diff --git a/pyproject.toml b/pyproject.toml
index 69b8f4dd..0cb13bb1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,11 +1,13 @@
[tool.black]
target-version = ['py37']
[tool.isort]
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
ensure_newline_before_comments = true
line_length = 88
force_sort_within_sections = true
+known_django = 'django,rest_framework'
+sections = 'FUTURE,STDLIB,THIRDPARTY,DJANGO,FIRSTPARTY,LOCALFOLDER'
diff --git a/swh/web/admin/deposit.py b/swh/web/admin/deposit.py
index 02e02bf2..ab773062 100644
--- a/swh/web/admin/deposit.py
+++ b/swh/web/admin/deposit.py
@@ -1,108 +1,109 @@
# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import requests
+from requests.auth import HTTPBasicAuth
+import sentry_sdk
+
from django.conf import settings
from django.contrib.admin.views.decorators import staff_member_required
from django.core.cache import cache
from django.core.paginator import Paginator
from django.http import JsonResponse
from django.shortcuts import render
-import requests
-from requests.auth import HTTPBasicAuth
-import sentry_sdk
from swh.web.admin.adminurls import admin_route
from swh.web.config import get_config
config = get_config()["deposit"]
@admin_route(r"deposit/", view_name="admin-deposit")
@staff_member_required(view_func=None, login_url=settings.LOGIN_URL)
def _admin_origin_save(request):
return render(request, "admin/deposit.html")
@admin_route(r"deposit/list/", view_name="admin-deposit-list")
@staff_member_required(view_func=None, login_url=settings.LOGIN_URL)
def _admin_deposit_list(request):
table_data = {}
table_data["draw"] = int(request.GET["draw"])
deposits_list_url = config["private_api_url"] + "deposits"
deposits_list_auth = HTTPBasicAuth(
config["private_api_user"], config["private_api_password"]
)
try:
nb_deposits = requests.get(
"%s?page_size=1" % deposits_list_url, auth=deposits_list_auth, timeout=30
).json()["count"]
deposits_data = cache.get("swh-deposit-list")
if not deposits_data or deposits_data["count"] != nb_deposits:
deposits_data = requests.get(
"%s?page_size=%s" % (deposits_list_url, nb_deposits),
auth=deposits_list_auth,
timeout=30,
).json()
cache.set("swh-deposit-list", deposits_data)
deposits = deposits_data["results"]
search_value = request.GET["search[value]"]
if search_value:
deposits = [
d
for d in deposits
if any(
search_value.lower() in val
for val in [str(v).lower() for v in d.values()]
)
]
exclude_pattern = request.GET.get("excludePattern")
if exclude_pattern:
deposits = [
d
for d in deposits
if all(
exclude_pattern.lower() not in val
for val in [str(v).lower() for v in d.values()]
)
]
column_order = request.GET["order[0][column]"]
field_order = request.GET["columns[%s][name]" % column_order]
order_dir = request.GET["order[0][dir]"]
deposits = sorted(deposits, key=lambda d: d[field_order] or "")
if order_dir == "desc":
deposits = list(reversed(deposits))
length = int(request.GET["length"])
page = int(request.GET["start"]) / length + 1
paginator = Paginator(deposits, length)
data = paginator.page(page).object_list
table_data["recordsTotal"] = deposits_data["count"]
table_data["recordsFiltered"] = len(deposits)
table_data["data"] = [
{
"id": d["id"],
"external_id": d["external_id"],
"reception_date": d["reception_date"],
"status": d["status"],
"status_detail": d["status_detail"],
"swhid": d["swh_id"],
"swhid_context": d["swh_id_context"],
}
for d in data
]
except Exception as exc:
sentry_sdk.capture_exception(exc)
table_data["error"] = (
"An error occurred while retrieving " "the list of deposits !"
)
return JsonResponse(table_data)
diff --git a/swh/web/api/apidoc.py b/swh/web/api/apidoc.py
index dbcb1d72..11921cd2 100644
--- a/swh/web/api/apidoc.py
+++ b/swh/web/api/apidoc.py
@@ -1,452 +1,453 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from collections import defaultdict
import functools
from functools import wraps
import os
import re
import textwrap
from typing import List
import docutils.nodes
import docutils.parsers.rst
import docutils.utils
-from rest_framework.decorators import api_view
import sentry_sdk
+from rest_framework.decorators import api_view
+
from swh.web.api.apiresponse import error_response, make_api_response
from swh.web.api.apiurls import APIUrls
from swh.web.common.utils import parse_rst
class _HTTPDomainDocVisitor(docutils.nodes.NodeVisitor):
"""
docutils visitor for walking on a parsed rst document containing sphinx
httpdomain roles. Its purpose is to extract relevant info regarding swh
api endpoints (for instance url arguments) from their docstring written
using sphinx httpdomain.
"""
# httpdomain roles we want to parse (based on sphinxcontrib.httpdomain 1.6)
parameter_roles = ("param", "parameter", "arg", "argument")
request_json_object_roles = ("reqjsonobj", "reqjson", "<jsonobj", "<json")
request_json_array_roles = ("reqjsonarr", "<jsonarr")
response_json_object_roles = ("resjsonobj", "resjson", ">jsonobj", ">json")
response_json_array_roles = ("resjsonarr", ">jsonarr")
query_parameter_roles = ("queryparameter", "queryparam", "qparam", "query")
request_header_roles = ("<header", "reqheader", "requestheader")
response_header_roles = (">header", "resheader", "responseheader")
status_code_roles = ("statuscode", "status", "code")
def __init__(self, document, data):
super().__init__(document)
self.data = data
self.args_set = set()
self.params_set = set()
self.inputs_set = set()
self.returns_set = set()
self.status_codes_set = set()
self.reqheaders_set = set()
self.resheaders_set = set()
self.field_list_visited = False
self.current_json_obj = None
def process_paragraph(self, par):
"""
Process extracted paragraph text before display.
Cleanup document model markups and transform the
paragraph into a valid raw rst string (as the apidoc
documentation transform rst to html when rendering).
"""
par = par.replace("\n", " ")
# keep emphasized, strong and literal text
par = par.replace("<emphasis>", "*")
par = par.replace("</emphasis>", "*")
par = par.replace("<strong>", "**")
par = par.replace("</strong>", "**")
par = par.replace("<literal>", "``")
par = par.replace("</literal>", "``")
# keep links to web pages
if "<reference" in par:
par = re.sub(
r'<reference name="(.*)" refuri="(.*)".*</reference>',
r"`\1 <\2>`_",
par,
)
# remove parsed document markups but keep rst links
par = re.sub(r"<[^<]+?>(?!`_)", "", par)
# api urls cleanup to generate valid links afterwards
subs_made = 1
while subs_made:
(par, subs_made) = re.subn(r"(:http:.*)(\(\w+\))", r"\1", par)
subs_made = 1
while subs_made:
(par, subs_made) = re.subn(r"(:http:.*)(\[.*\])", r"\1", par)
par = par.replace("//", "/")
# transform references to api endpoints doc into valid rst links
par = re.sub(":http:get:`([^,`]*)`", r"`\1 <\1doc/>`_", par)
# transform references to some elements into bold text
par = re.sub(":http:header:`(.*)`", r"**\1**", par)
par = re.sub(":func:`(.*)`", r"**\1**", par)
return par
def visit_field_list(self, node):
"""
Visit parsed rst field lists to extract relevant info
regarding api endpoint.
"""
self.field_list_visited = True
for child in node.traverse():
# get the parsed field name
if isinstance(child, docutils.nodes.field_name):
field_name = child.astext()
# parse field text
elif isinstance(child, docutils.nodes.paragraph):
text = self.process_paragraph(str(child))
field_data = field_name.split(" ")
# Parameters
if field_data[0] in self.parameter_roles:
if field_data[2] not in self.args_set:
self.data["args"].append(
{"name": field_data[2], "type": field_data[1], "doc": text}
)
self.args_set.add(field_data[2])
# Query Parameters
if field_data[0] in self.query_parameter_roles:
if field_data[2] not in self.params_set:
self.data["params"].append(
{"name": field_data[2], "type": field_data[1], "doc": text}
)
self.params_set.add(field_data[2])
# Request data type
if (
field_data[0] in self.request_json_array_roles
or field_data[0] in self.request_json_object_roles
):
# array
if field_data[0] in self.request_json_array_roles:
self.data["input_type"] = "array"
# object
else:
self.data["input_type"] = "object"
# input object field
if field_data[2] not in self.inputs_set:
self.data["inputs"].append(
{"name": field_data[2], "type": field_data[1], "doc": text}
)
self.inputs_set.add(field_data[2])
self.current_json_obj = self.data["inputs"][-1]
# Response type
if (
field_data[0] in self.response_json_array_roles
or field_data[0] in self.response_json_object_roles
):
# array
if field_data[0] in self.response_json_array_roles:
self.data["return_type"] = "array"
# object
else:
self.data["return_type"] = "object"
# returned object field
if field_data[2] not in self.returns_set:
self.data["returns"].append(
{"name": field_data[2], "type": field_data[1], "doc": text}
)
self.returns_set.add(field_data[2])
self.current_json_obj = self.data["returns"][-1]
# Status Codes
if field_data[0] in self.status_code_roles:
if field_data[1] not in self.status_codes_set:
self.data["status_codes"].append(
{"code": field_data[1], "doc": text}
)
self.status_codes_set.add(field_data[1])
# Request Headers
if field_data[0] in self.request_header_roles:
if field_data[1] not in self.reqheaders_set:
self.data["reqheaders"].append(
{"name": field_data[1], "doc": text}
)
self.reqheaders_set.add(field_data[1])
# Response Headers
if field_data[0] in self.response_header_roles:
if field_data[1] not in self.resheaders_set:
resheader = {"name": field_data[1], "doc": text}
self.data["resheaders"].append(resheader)
self.resheaders_set.add(field_data[1])
if (
resheader["name"] == "Content-Type"
and resheader["doc"] == "application/octet-stream"
):
self.data["return_type"] = "octet stream"
def visit_paragraph(self, node):
"""
Visit relevant paragraphs to parse
"""
# only parsed top level paragraphs
if isinstance(node.parent, docutils.nodes.block_quote):
text = self.process_paragraph(str(node))
# endpoint description
if not text.startswith("**") and text not in self.data["description"]:
self.data["description"] += "\n\n" if self.data["description"] else ""
self.data["description"] += text
def visit_literal_block(self, node):
"""
Visit literal blocks
"""
text = node.astext()
# literal block in endpoint description
if not self.field_list_visited:
self.data["description"] += ":\n\n%s\n" % textwrap.indent(text, "\t")
# extract example url
if ":swh_web_api:" in text:
self.data["examples"].append("/api/1/" + re.sub(".*`(.*)`.*", r"\1", text))
def visit_bullet_list(self, node):
# bullet list in endpoint description
if not self.field_list_visited:
self.data["description"] += "\n\n"
for child in node.traverse():
# process list item
if isinstance(child, docutils.nodes.paragraph):
line_text = self.process_paragraph(str(child))
self.data["description"] += "\t* %s\n" % line_text
elif self.current_json_obj:
self.current_json_obj["doc"] += "\n\n"
for child in node.traverse():
# process list item
if isinstance(child, docutils.nodes.paragraph):
line_text = self.process_paragraph(str(child))
self.current_json_obj["doc"] += "\t\t* %s\n" % line_text
self.current_json_obj = None
def visit_warning(self, node):
text = self.process_paragraph(str(node))
rst_warning = "\n\n.. warning::\n%s\n" % textwrap.indent(text, "\t")
if rst_warning not in self.data["description"]:
self.data["description"] += rst_warning
def unknown_visit(self, node):
pass
def unknown_departure(self, node):
pass
def _parse_httpdomain_doc(doc, data):
doc_lines = doc.split("\n")
doc_lines_filtered = []
urls = defaultdict(list)
default_http_methods = ["HEAD", "OPTIONS"]
# httpdomain is a sphinx extension that is unknown to docutils but
# fortunately we can still parse its directives' content,
# so remove lines with httpdomain directives before executing the
# rst parser from docutils
for doc_line in doc_lines:
if ".. http" not in doc_line:
doc_lines_filtered.append(doc_line)
else:
url = doc_line[doc_line.find("/") :]
# emphasize url arguments for html rendering
url = re.sub(r"\((\w+)\)", r" **\(\1\)** ", url)
method = re.search(r"http:(\w+)::", doc_line).group(1)
urls[url].append(method.upper())
for url, methods in urls.items():
data["urls"].append({"rule": url, "methods": methods + default_http_methods})
# parse the rst docstring and do not print system messages about
# unknown httpdomain roles
document = parse_rst("\n".join(doc_lines_filtered), report_level=5)
# remove the system_message nodes from the parsed document
for node in document.traverse(docutils.nodes.system_message):
node.parent.remove(node)
# visit the document nodes to extract relevant endpoint info
visitor = _HTTPDomainDocVisitor(document, data)
document.walkabout(visitor)
class APIDocException(Exception):
"""
Custom exception to signal errors in the use of the APIDoc decorators
"""
def api_doc(
route: str,
noargs: bool = False,
tags: List[str] = [],
handle_response: bool = False,
api_version: str = "1",
):
"""
Decorator for an API endpoint implementation used to generate a dedicated
view displaying its HTML documentation.
The documentation will be generated from the endpoint docstring based on
sphinxcontrib-httpdomain format.
Args:
route: documentation page's route
noargs: set to True if the route has no arguments, and its
result should be displayed anytime its documentation
is requested. Default to False
tags: Further information on api endpoints. Two values are
possibly expected:
* hidden: remove the entry points from the listing
* upcoming: display the entry point but it is not followable
handle_response: indicate if the decorated function takes
care of creating the HTTP response or delegates that task to the
apiresponse module
api_version: api version string
"""
tags_set = set(tags)
# @api_doc() Decorator call
def decorator(f):
# if the route is not hidden, add it to the index
if "hidden" not in tags_set:
doc_data = get_doc_data(f, route, noargs)
doc_desc = doc_data["description"]
first_dot_pos = doc_desc.find(".")
APIUrls.add_doc_route(
route,
doc_desc[: first_dot_pos + 1],
noargs=noargs,
api_version=api_version,
tags=tags_set,
)
# create a dedicated view to display endpoint HTML doc
@api_view(["GET", "HEAD"])
@wraps(f)
def doc_view(request):
doc_data = get_doc_data(f, route, noargs)
return make_api_response(request, None, doc_data)
route_name = "%s-doc" % route[1:-1].replace("/", "-")
urlpattern = f"^{api_version}{route}doc/$"
view_name = "api-%s-%s" % (api_version, route_name)
APIUrls.add_url_pattern(urlpattern, doc_view, view_name)
@wraps(f)
def documented_view(request, **kwargs):
doc_data = get_doc_data(f, route, noargs)
try:
response = f(request, **kwargs)
except Exception as exc:
sentry_sdk.capture_exception(exc)
return error_response(request, exc, doc_data)
if handle_response:
return response
else:
return make_api_response(request, response, doc_data)
return documented_view
return decorator
@functools.lru_cache(maxsize=32)
def get_doc_data(f, route, noargs):
"""
Build documentation data for the decorated api endpoint function
"""
data = {
"description": "",
"response_data": None,
"urls": [],
"args": [],
"params": [],
"input_type": "",
"inputs": [],
"resheaders": [],
"reqheaders": [],
"return_type": "",
"returns": [],
"status_codes": [],
"examples": [],
"route": route,
"noargs": noargs,
}
if not f.__doc__:
raise APIDocException(
"apidoc: expected a docstring" " for function %s" % (f.__name__,)
)
# use raw docstring as endpoint documentation if sphinx
# httpdomain is not used
if ".. http" not in f.__doc__:
data["description"] = f.__doc__
# else parse the sphinx httpdomain docstring with docutils
# (except when building the swh-web documentation through autodoc
# sphinx extension, not needed and raise errors with sphinx >= 1.7)
elif "SWH_WEB_DOC_BUILD" not in os.environ:
_parse_httpdomain_doc(f.__doc__, data)
# process input/returned object info for nicer html display
inputs_list = ""
returns_list = ""
for inp in data["inputs"]:
# special case for array of non object type, for instance
# :<jsonarr string -: an array of string
if inp["name"] != "-":
inputs_list += "\t* **%s (%s)**: %s\n" % (
inp["name"],
inp["type"],
inp["doc"],
)
for ret in data["returns"]:
# special case for array of non object type, for instance
# :>jsonarr string -: an array of string
if ret["name"] != "-":
returns_list += "\t* **%s (%s)**: %s\n" % (
ret["name"],
ret["type"],
ret["doc"],
)
data["inputs_list"] = inputs_list
data["returns_list"] = returns_list
return data
DOC_COMMON_HEADERS = """
:reqheader Accept: the requested response content type,
either ``application/json`` (default) or ``application/yaml``
:resheader Content-Type: this depends on :http:header:`Accept`
header of request"""
DOC_RESHEADER_LINK = """
:resheader Link: indicates that a subsequent result page is
available and contains the url pointing to it
"""
DEFAULT_SUBSTITUTIONS = {
"common_headers": DOC_COMMON_HEADERS,
"resheader_link": DOC_RESHEADER_LINK,
}
def format_docstring(**substitutions):
def decorator(f):
f.__doc__ = f.__doc__.format(**{**DEFAULT_SUBSTITUTIONS, **substitutions})
return f
return decorator
diff --git a/swh/web/api/renderers.py b/swh/web/api/renderers.py
index 45dd36a0..bd1f5af9 100644
--- a/swh/web/api/renderers.py
+++ b/swh/web/api/renderers.py
@@ -1,36 +1,37 @@
# Copyright (C) 2017-2018 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
-from rest_framework import renderers
import yaml
+from rest_framework import renderers
+
class YAMLRenderer(renderers.BaseRenderer):
"""
Renderer which serializes to YAML.
"""
media_type = "application/yaml"
format = "yaml"
charset = "utf-8"
ensure_ascii = False
default_flow_style = False
def render(self, data, accepted_media_type=None, renderer_context=None):
"""
Renders `data` into serialized YAML.
"""
assert yaml, "YAMLRenderer requires pyyaml to be installed"
if data is None:
return ""
return yaml.dump(
data,
stream=None,
encoding=self.charset,
allow_unicode=not self.ensure_ascii,
default_flow_style=self.default_flow_style,
)
diff --git a/swh/web/api/throttling.py b/swh/web/api/throttling.py
index cfc09b44..4a14b98a 100644
--- a/swh/web/api/throttling.py
+++ b/swh/web/api/throttling.py
@@ -1,167 +1,168 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network
from typing import Callable, List, TypeVar, Union
+import sentry_sdk
+
from django.core.exceptions import ImproperlyConfigured
import rest_framework
from rest_framework.throttling import ScopedRateThrottle
-import sentry_sdk
from swh.web.config import get_config
APIView = TypeVar("APIView", bound="rest_framework.views.APIView")
Request = rest_framework.request.Request
API_THROTTLING_EXEMPTED_PERM = "swh.web.api.throttling_exempted"
class SwhWebRateThrottle(ScopedRateThrottle):
"""Custom request rate limiter for DRF enabling to exempt
specific networks specified in swh-web configuration.
Requests are grouped into scopes. It enables to apply different
requests rate limiting based on the scope name but also the
input HTTP request types.
To associate a scope to requests, one must add a 'throttle_scope'
attribute when using a class based view, or call the 'throttle_scope'
decorator when using a function based view. By default, requests
do not have an associated scope and are not rate limited.
Rate limiting can also be configured according to the type
of the input HTTP requests for fine grained tuning.
For instance, the following YAML configuration section sets a rate of:
- 1 per minute for POST requests
- 60 per minute for other request types
for the 'swh_api' scope while exempting those coming from the
127.0.0.0/8 ip network.
.. code-block:: yaml
throttling:
scopes:
swh_api:
limiter_rate:
default: 60/m
POST: 1/m
exempted_networks:
- 127.0.0.0/8
"""
scope = None
def __init__(self):
super().__init__()
self.exempted_networks = None
def get_exempted_networks(
self, scope_name: str
) -> List[Union[IPv4Network, IPv6Network]]:
if not self.exempted_networks:
scopes = get_config()["throttling"]["scopes"]
scope = scopes.get(scope_name)
if scope:
networks = scope.get("exempted_networks")
if networks:
self.exempted_networks = [
ip_network(network) for network in networks
]
return self.exempted_networks
def allow_request(self, request: Request, view: APIView) -> bool:
# no throttling for staff users
if request.user.is_authenticated and (
request.user.is_staff or request.user.has_perm(API_THROTTLING_EXEMPTED_PERM)
):
return True
# class based view case
if not self.scope:
default_scope = getattr(view, self.scope_attr, None)
request_allowed = None
if default_scope is not None:
# check if there is a specific rate limiting associated
# to the request type
assert request.method is not None
request_scope = f"{default_scope}_{request.method.lower()}"
setattr(view, self.scope_attr, request_scope)
try:
request_allowed = super().allow_request(request, view)
# use default rate limiting otherwise
except ImproperlyConfigured as exc:
sentry_sdk.capture_exception(exc)
setattr(view, self.scope_attr, default_scope)
if request_allowed is None:
request_allowed = super().allow_request(request, view)
# function based view case
else:
default_scope = self.scope
# check if there is a specific rate limiting associated
# to the request type
self.scope = default_scope + "_" + request.method.lower()
try:
self.rate = self.get_rate()
# use default rate limiting otherwise
except ImproperlyConfigured as exc:
sentry_sdk.capture_exception(exc)
self.scope = default_scope
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
request_allowed = super(ScopedRateThrottle, self).allow_request(
request, view
)
self.scope = default_scope
exempted_networks = self.get_exempted_networks(default_scope)
exempted_ip = False
if exempted_networks:
remote_address = ip_address(self.get_ident(request))
exempted_ip = any(
remote_address in network for network in exempted_networks
)
request_allowed = exempted_ip or request_allowed
# set throttling related data in the request metadata
# in order for the ThrottlingHeadersMiddleware to
# add X-RateLimit-* headers in the HTTP response
if not exempted_ip and hasattr(self, "history"):
hit_count = len(self.history)
request.META["RateLimit-Limit"] = self.num_requests
request.META["RateLimit-Remaining"] = self.num_requests - hit_count
wait = self.wait()
if wait is not None:
request.META["RateLimit-Reset"] = int(self.now + wait)
return request_allowed
def throttle_scope(scope: str) -> Callable[..., APIView]:
"""Decorator that allows the throttle scope of a DRF
function based view to be set::
@api_view(['GET', ])
@throttle_scope('scope')
def view(request):
...
"""
def decorator(func: APIView) -> APIView:
SwhScopeRateThrottle = type(
"CustomScopeRateThrottle", (SwhWebRateThrottle,), {"scope": scope}
)
func.throttle_classes = (SwhScopeRateThrottle,)
return func
return decorator
diff --git a/swh/web/api/views/utils.py b/swh/web/api/views/utils.py
index 7cb3ef57..901ef605 100644
--- a/swh/web/api/views/utils.py
+++ b/swh/web/api/views/utils.py
@@ -1,92 +1,93 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from types import GeneratorType
from typing import Any, Callable, Dict, Mapping, Optional
+from typing_extensions import Protocol
+
from django.http import HttpRequest
from rest_framework.decorators import api_view
from rest_framework.response import Response
-from typing_extensions import Protocol
from swh.web.api.apiurls import APIUrls, api_route
from swh.web.common.exc import NotFoundExc
class EnrichFunction(Protocol):
def __call__(
self, input: Mapping[str, str], request: Optional[HttpRequest]
) -> Dict[str, str]:
...
def api_lookup(
lookup_fn: Callable[..., Any],
*args: Any,
notfound_msg: Optional[str] = "Object not found",
enrich_fn: Optional[EnrichFunction] = None,
request: Optional[HttpRequest] = None,
):
r"""
Capture a redundant behavior of:
- looking up the backend with a criteria (be it an identifier or
checksum) passed to the function lookup_fn
- if nothing is found, raise an NotFoundExc exception with error
message notfound_msg.
- Otherwise if something is returned:
- either as list, map or generator, map the enrich_fn function to
it and return the resulting data structure as list.
- either as dict and pass to enrich_fn and return the dict
enriched.
Args:
- lookup_fn: function expects one criteria and optional supplementary
\*args.
- \*args: supplementary arguments to pass to lookup_fn.
- notfound_msg: if nothing matching the criteria is found,
raise NotFoundExc with this error message.
- enrich_fn: Function to use to enrich the result returned by
lookup_fn. Default to the identity function if not provided.
- request: Input HTTP request that will be provided as parameter
to enrich_fn.
Raises:
NotFoundExp or whatever `lookup_fn` raises.
"""
def _enrich_fn_noop(x, request):
return x
if enrich_fn is None:
enrich_fn = _enrich_fn_noop
res = lookup_fn(*args)
if res is None:
raise NotFoundExc(notfound_msg)
if isinstance(res, (list, GeneratorType)) or type(res) == map:
return [enrich_fn(x, request=request) for x in res]
return enrich_fn(res, request=request)
@api_view(["GET", "HEAD"])
def api_home(request):
return Response({}, template_name="api/api.html")
APIUrls.add_url_pattern(r"^$", api_home, view_name="api-1-homepage")
@api_route(r"/", "api-1-endpoints")
def api_endpoints(request):
"""Display the list of opened api endpoints.
"""
routes = APIUrls.get_app_endpoints().copy()
for route, doc in routes.items():
doc["doc_intro"] = doc["docstring"].split("\n\n")[0]
# Return a list of routes with consistent ordering
env = {"doc_routes": sorted(routes.items())}
return Response(env, template_name="api/endpoints.html")
diff --git a/swh/web/auth/backends.py b/swh/web/auth/backends.py
index 814f4001..0a83017e 100644
--- a/swh/web/auth/backends.py
+++ b/swh/web/auth/backends.py
@@ -1,177 +1,178 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime, timedelta
import hashlib
from typing import Any, Dict, Optional
+import sentry_sdk
+
from django.core.cache import cache
from django.http import HttpRequest
from django.utils import timezone
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed, ValidationError
-import sentry_sdk
from swh.web.auth.keycloak import KeycloakOpenIDConnect
from swh.web.auth.models import OIDCUser
from swh.web.auth.utils import get_oidc_client
# OpenID Connect client to communicate with Keycloak server
_oidc_client: KeycloakOpenIDConnect = get_oidc_client()
def _oidc_user_from_decoded_token(decoded_token: Dict[str, Any]) -> OIDCUser:
# compute an integer user identifier for Django User model
# by concatenating all groups of the UUID4 user identifier
# generated by Keycloak and converting it from hex to decimal
user_id = int("".join(decoded_token["sub"].split("-")), 16)
# create a Django user that will not be saved to database
user = OIDCUser(
id=user_id,
username=decoded_token["preferred_username"],
password="",
first_name=decoded_token["given_name"],
last_name=decoded_token["family_name"],
email=decoded_token["email"],
)
# set is_staff user property based on groups
if "groups" in decoded_token:
user.is_staff = "/staff" in decoded_token["groups"]
# extract user permissions if any
resource_access = decoded_token.get("resource_access", {})
client_resource_access = resource_access.get(_oidc_client.client_id, {})
user.permissions = set(client_resource_access.get("roles", []))
# add user sub to custom User proxy model
user.sub = decoded_token["sub"]
return user
def _oidc_user_from_profile(oidc_profile: Dict[str, Any]) -> OIDCUser:
# decode JWT token
decoded_token = _oidc_client.decode_token(oidc_profile["access_token"])
# create OIDCUser from decoded token
user = _oidc_user_from_decoded_token(decoded_token)
# get authentication init datetime
auth_datetime = datetime.fromtimestamp(decoded_token["auth_time"])
exp_datetime = datetime.fromtimestamp(decoded_token["exp"])
# compute OIDC tokens expiration date
oidc_profile["expires_at"] = exp_datetime
oidc_profile["refresh_expires_at"] = auth_datetime + timedelta(
seconds=oidc_profile["refresh_expires_in"]
)
# add OIDC profile data to custom User proxy model
for key, val in oidc_profile.items():
if hasattr(user, key):
setattr(user, key, val)
return user
class OIDCAuthorizationCodePKCEBackend:
def authenticate(
self, request: HttpRequest, code: str, code_verifier: str, redirect_uri: str
) -> Optional[OIDCUser]:
user = None
try:
# try to authenticate user with OIDC PKCE authorization code flow
oidc_profile = _oidc_client.authorization_code(
code, redirect_uri, code_verifier=code_verifier
)
# create Django user
user = _oidc_user_from_profile(oidc_profile)
# set cache key TTL as access token expiration time
assert user.expires_at
ttl = int(user.expires_at.timestamp() - timezone.now().timestamp())
# save oidc_profile in cache
cache.set(f"oidc_user_{user.id}", oidc_profile, timeout=max(0, ttl))
except Exception as e:
sentry_sdk.capture_exception(e)
return user
def get_user(self, user_id: int) -> Optional[OIDCUser]:
# get oidc profile from cache
oidc_profile = cache.get(f"oidc_user_{user_id}")
if oidc_profile:
try:
user = _oidc_user_from_profile(oidc_profile)
# restore auth backend
setattr(user, "backend", f"{__name__}.{self.__class__.__name__}")
return user
except Exception as e:
sentry_sdk.capture_exception(e)
return None
else:
return None
class OIDCBearerTokenAuthentication(BaseAuthentication):
def authenticate(self, request):
auth_header = request.META.get("HTTP_AUTHORIZATION")
if auth_header is None:
return None
try:
auth_type, refresh_token = auth_header.split(" ", 1)
except ValueError:
raise AuthenticationFailed("Invalid HTTP authorization header format")
if auth_type != "Bearer":
raise AuthenticationFailed(
(f"Invalid or unsupported HTTP authorization" f" type ({auth_type}).")
)
try:
# compute a cache key from the token that does not exceed
# memcached key size limit
hasher = hashlib.sha1()
hasher.update(refresh_token.encode("ascii"))
cache_key = f"api_token_{hasher.hexdigest()}"
# check if an access token is cached
access_token = cache.get(cache_key)
# attempt to decode access token
try:
decoded_token = _oidc_client.decode_token(access_token)
except Exception:
# access token is None or it has expired
decoded_token = None
if access_token is None or decoded_token is None:
# get a new access token from authentication provider
access_token = _oidc_client.refresh_token(refresh_token)["access_token"]
# decode access token
decoded_token = _oidc_client.decode_token(access_token)
# compute access token expiration
exp = datetime.fromtimestamp(decoded_token["exp"])
ttl = int(exp.timestamp() - timezone.now().timestamp())
# save access token in cache while it is valid
cache.set(cache_key, access_token, timeout=max(0, ttl))
# create Django user
user = _oidc_user_from_decoded_token(decoded_token)
except UnicodeEncodeError as e:
sentry_sdk.capture_exception(e)
raise ValidationError("Invalid bearer token")
except Exception as e:
sentry_sdk.capture_exception(e)
raise AuthenticationFailed(str(e))
return user, None
diff --git a/swh/web/browse/utils.py b/swh/web/browse/utils.py
index 2fecef19..8d1f80d8 100644
--- a/swh/web/browse/utils.py
+++ b/swh/web/browse/utils.py
@@ -1,741 +1,742 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import base64
import stat
import textwrap
from threading import Lock
+import magic
+import sentry_sdk
+
from django.core.cache import cache
from django.utils.html import escape
from django.utils.safestring import mark_safe
-import magic
-import sentry_sdk
from swh.web.common import highlightjs, service
from swh.web.common.exc import http_status_code_message
from swh.web.common.utils import (
browsers_supported_image_mimes,
format_utc_iso_date,
reverse,
rst_to_html,
)
from swh.web.config import get_config
def get_directory_entries(sha1_git):
"""Function that retrieves the content of a directory
from the archive.
The directories entries are first sorted in lexicographical order.
Sub-directories and regular files are then extracted.
Args:
sha1_git: sha1_git identifier of the directory
Returns:
A tuple whose first member corresponds to the sub-directories list
and second member the regular files list
Raises:
NotFoundExc if the directory is not found
"""
cache_entry_id = "directory_entries_%s" % sha1_git
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry
entries = list(service.lookup_directory(sha1_git))
for e in entries:
e["perms"] = stat.filemode(e["perms"])
if e["type"] == "rev":
# modify dir entry name to explicitly show it points
# to a revision
e["name"] = "%s @ %s" % (e["name"], e["target"][:7])
dirs = [e for e in entries if e["type"] in ("dir", "rev")]
files = [e for e in entries if e["type"] == "file"]
dirs = sorted(dirs, key=lambda d: d["name"])
files = sorted(files, key=lambda f: f["name"])
cache.set(cache_entry_id, (dirs, files))
return dirs, files
_lock = Lock()
def get_mimetype_and_encoding_for_content(content):
"""Function that returns the mime type and the encoding associated to
a content buffer using the magic module under the hood.
Args:
content (bytes): a content buffer
Returns:
A tuple (mimetype, encoding), for instance ('text/plain', 'us-ascii'),
associated to the provided content.
"""
m = magic.Magic(mime=True, mime_encoding=True)
mime_encoding = m.from_buffer(content)
mime_type, encoding = mime_encoding.split(";")
encoding = encoding.replace(" charset=", "")
return mime_type, encoding
# maximum authorized content size in bytes for HTML display
# with code highlighting
content_display_max_size = get_config()["content_display_max_size"]
def _re_encode_content(mimetype, encoding, content_data):
# encode textual content to utf-8 if needed
if mimetype.startswith("text/"):
# probably a malformed UTF-8 content, re-encode it
# by replacing invalid chars with a substitution one
if encoding == "unknown-8bit":
content_data = content_data.decode("utf-8", "replace").encode("utf-8")
elif encoding not in ["utf-8", "binary"]:
content_data = content_data.decode(encoding, "replace").encode("utf-8")
elif mimetype.startswith("application/octet-stream"):
# file may detect a text content as binary
# so try to decode it for display
encodings = ["us-ascii", "utf-8"]
encodings += ["iso-8859-%s" % i for i in range(1, 17)]
for enc in encodings:
try:
content_data = content_data.decode(enc).encode("utf-8")
except Exception as exc:
sentry_sdk.capture_exception(exc)
else:
# ensure display in content view
encoding = enc
mimetype = "text/plain"
break
return mimetype, encoding, content_data
def request_content(
query_string,
max_size=content_display_max_size,
raise_if_unavailable=True,
re_encode=True,
):
"""Function that retrieves a content from the archive.
Raw bytes content is first retrieved, then the content mime type.
If the mime type is not stored in the archive, it will be computed
using Python magic module.
Args:
query_string: a string of the form "[ALGO_HASH:]HASH" where
optional ALGO_HASH can be either ``sha1``, ``sha1_git``,
``sha256``, or ``blake2s256`` (default to ``sha1``) and HASH
the hexadecimal representation of the hash value
max_size: the maximum size for a content to retrieve (default to 1MB,
no size limit if None)
Returns:
A tuple whose first member corresponds to the content raw bytes
and second member the content mime type
Raises:
NotFoundExc if the content is not found
"""
content_data = service.lookup_content(query_string)
filetype = None
language = None
license = None
# requests to the indexer db may fail so properly handle
# those cases in order to avoid content display errors
try:
filetype = service.lookup_content_filetype(query_string)
language = service.lookup_content_language(query_string)
license = service.lookup_content_license(query_string)
except Exception as exc:
sentry_sdk.capture_exception(exc)
mimetype = "unknown"
encoding = "unknown"
if filetype:
mimetype = filetype["mimetype"]
encoding = filetype["encoding"]
# workaround when encountering corrupted data due to implicit
# conversion from bytea to text in the indexer db (see T818)
# TODO: Remove that code when all data have been correctly converted
if mimetype.startswith("\\"):
filetype = None
content_data["error_code"] = 200
content_data["error_message"] = ""
content_data["error_description"] = ""
if not max_size or content_data["length"] < max_size:
try:
content_raw = service.lookup_content_raw(query_string)
except Exception as exc:
if raise_if_unavailable:
raise exc
else:
sentry_sdk.capture_exception(exc)
content_data["raw_data"] = None
content_data["error_code"] = 404
content_data["error_description"] = (
"The bytes of the content are currently not available "
"in the archive."
)
content_data["error_message"] = http_status_code_message[
content_data["error_code"]
]
else:
content_data["raw_data"] = content_raw["data"]
if not filetype:
mimetype, encoding = get_mimetype_and_encoding_for_content(
content_data["raw_data"]
)
if re_encode:
mimetype, encoding, raw_data = _re_encode_content(
mimetype, encoding, content_data["raw_data"]
)
content_data["raw_data"] = raw_data
else:
content_data["raw_data"] = None
content_data["mimetype"] = mimetype
content_data["encoding"] = encoding
if language:
content_data["language"] = language["lang"]
else:
content_data["language"] = "not detected"
if license:
content_data["licenses"] = ", ".join(license["facts"][0]["licenses"])
else:
content_data["licenses"] = "not detected"
return content_data
def prepare_content_for_display(content_data, mime_type, path):
"""Function that prepares a content for HTML display.
The function tries to associate a programming language to a
content in order to perform syntax highlighting client-side
using highlightjs. The language is determined using either
the content filename or its mime type.
If the mime type corresponds to an image format supported
by web browsers, the content will be encoded in base64
for displaying the image.
Args:
content_data (bytes): raw bytes of the content
mime_type (string): mime type of the content
path (string): path of the content including filename
Returns:
A dict containing the content bytes (possibly different from the one
provided as parameter if it is an image) under the key 'content_data
and the corresponding highlightjs language class under the
key 'language'.
"""
language = highlightjs.get_hljs_language_from_filename(path)
if not language:
language = highlightjs.get_hljs_language_from_mime_type(mime_type)
if not language:
language = "nohighlight"
elif mime_type.startswith("application/"):
mime_type = mime_type.replace("application/", "text/")
if mime_type.startswith("image/"):
if mime_type in browsers_supported_image_mimes:
content_data = base64.b64encode(content_data).decode("ascii")
if mime_type.startswith("image/svg"):
mime_type = "image/svg+xml"
if mime_type.startswith("text/"):
content_data = content_data.decode("utf-8", errors="replace")
return {"content_data": content_data, "language": language, "mimetype": mime_type}
def gen_link(url, link_text=None, link_attrs=None):
"""
Utility function for generating an HTML link to insert
in Django templates.
Args:
url (str): an url
link_text (str): optional text for the produced link,
if not provided the url will be used
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="url">link_text</a>'
"""
attrs = " "
if link_attrs:
for k, v in link_attrs.items():
attrs += '%s="%s" ' % (k, v)
if not link_text:
link_text = url
link = '<a%shref="%s">%s</a>' % (attrs, escape(url), escape(link_text))
return mark_safe(link)
def _snapshot_context_query_params(snapshot_context):
query_params = {}
if not snapshot_context:
return query_params
if snapshot_context and snapshot_context["origin_info"]:
origin_info = snapshot_context["origin_info"]
snp_query_params = snapshot_context["query_params"]
query_params = {"origin_url": origin_info["url"]}
if "timestamp" in snp_query_params:
query_params["timestamp"] = snp_query_params["timestamp"]
if "visit_id" in snp_query_params:
query_params["visit_id"] = snp_query_params["visit_id"]
if "snapshot" in snp_query_params and "visit_id" not in query_params:
query_params["snapshot"] = snp_query_params["snapshot"]
elif snapshot_context:
query_params = {"snapshot": snapshot_context["snapshot_id"]}
if snapshot_context["release"]:
query_params["release"] = snapshot_context["release"]
elif snapshot_context["branch"] and snapshot_context["branch"] not in (
"HEAD",
snapshot_context["revision_id"],
):
query_params["branch"] = snapshot_context["branch"]
elif snapshot_context["revision_id"]:
query_params["revision"] = snapshot_context["revision_id"]
return query_params
def gen_revision_url(revision_id, snapshot_context=None):
"""
Utility function for generating an url to a revision.
Args:
revision_id (str): a revision id
snapshot_context (dict): if provided, generate snapshot-dependent
browsing url
Returns:
str: The url to browse the revision
"""
query_params = _snapshot_context_query_params(snapshot_context)
query_params.pop("revision", None)
return reverse(
"browse-revision", url_args={"sha1_git": revision_id}, query_params=query_params
)
def gen_revision_link(
revision_id,
shorten_id=False,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a revision HTML view
to insert in Django templates.
Args:
revision_id (str): a revision id
shorten_id (boolean): whether to shorten the revision id to 7
characters for the link text
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
str: An HTML link in the form '<a href="revision_url">revision_id</a>'
"""
if not revision_id:
return None
revision_url = gen_revision_url(revision_id, snapshot_context)
if shorten_id:
return gen_link(revision_url, revision_id[:7], link_attrs)
else:
if not link_text:
link_text = revision_id
return gen_link(revision_url, link_text, link_attrs)
def gen_directory_link(
sha1_git,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a directory HTML view
to insert in Django templates.
Args:
sha1_git (str): directory identifier
link_text (str): optional text for the generated link
(the directory id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="directory_view_url">link_text</a>'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
directory_url = reverse(
"browse-directory", url_args={"sha1_git": sha1_git}, query_params=query_params
)
if not link_text:
link_text = sha1_git
return gen_link(directory_url, link_text, link_attrs)
def gen_snapshot_link(
snapshot_id,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a snapshot HTML view
to insert in Django templates.
Args:
snapshot_id (str): snapshot identifier
link_text (str): optional text for the generated link
(the snapshot id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="snapshot_view_url">link_text</a>'
"""
query_params = _snapshot_context_query_params(snapshot_context)
snapshot_url = reverse(
"browse-snapshot",
url_args={"snapshot_id": snapshot_id},
query_params=query_params,
)
if not link_text:
link_text = snapshot_id
return gen_link(snapshot_url, link_text, link_attrs)
def gen_content_link(
sha1_git,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a content HTML view
to insert in Django templates.
Args:
sha1_git (str): content identifier
link_text (str): optional text for the generated link
(the content sha1_git will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="content_view_url">link_text</a>'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
content_url = reverse(
"browse-content",
url_args={"query_string": "sha1_git:" + sha1_git},
query_params=query_params,
)
if not link_text:
link_text = sha1_git
return gen_link(content_url, link_text, link_attrs)
def get_revision_log_url(revision_id, snapshot_context=None):
"""
Utility function for getting the URL for a revision log HTML view
(possibly in the context of an origin).
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
Returns:
The revision log view URL
"""
query_params = {}
if snapshot_context:
query_params = _snapshot_context_query_params(snapshot_context)
query_params["revision"] = revision_id
if snapshot_context and snapshot_context["origin_info"]:
revision_log_url = reverse("browse-origin-log", query_params=query_params)
elif snapshot_context:
url_args = {"snapshot_id": snapshot_context["snapshot_id"]}
del query_params["snapshot"]
revision_log_url = reverse(
"browse-snapshot-log", url_args=url_args, query_params=query_params
)
else:
revision_log_url = reverse(
"browse-revision-log", url_args={"sha1_git": revision_id}
)
return revision_log_url
def gen_revision_log_link(
revision_id,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a revision log HTML view
(possibly in the context of an origin) to insert in Django templates.
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text to use for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form
'<a href="revision_log_view_url">link_text</a>'
"""
if not revision_id:
return None
revision_log_url = get_revision_log_url(revision_id, snapshot_context)
if not link_text:
link_text = revision_id
return gen_link(revision_log_url, link_text, link_attrs)
def gen_person_mail_link(person, link_text=None):
"""
Utility function for generating a mail link to a person to insert
in Django templates.
Args:
person (dict): dictionary containing person data
(*name*, *email*, *fullname*)
link_text (str): optional text to use for the generated mail link
(the person name will be used by default)
Returns:
str: A mail link to the person or the person name if no email is
present in person data
"""
person_name = person["name"] or person["fullname"] or "None"
if link_text is None:
link_text = person_name
person_email = person["email"] if person["email"] else None
if person_email is None and "@" in person_name and " " not in person_name:
person_email = person_name
if person_email:
return gen_link(url="mailto:%s" % person_email, link_text=link_text)
else:
return person_name
def gen_release_link(
sha1_git,
snapshot_context=None,
link_text="Browse",
link_attrs={"class": "btn btn-default btn-sm", "role": "button"},
):
"""
Utility function for generating a link to a release HTML view
to insert in Django templates.
Args:
sha1_git (str): release identifier
link_text (str): optional text for the generated link
(the release id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="release_view_url">link_text</a>'
"""
query_params = _snapshot_context_query_params(snapshot_context)
release_url = reverse(
"browse-release", url_args={"sha1_git": sha1_git}, query_params=query_params
)
if not link_text:
link_text = sha1_git
return gen_link(release_url, link_text, link_attrs)
def format_log_entries(revision_log, per_page, snapshot_context=None):
"""
Utility functions that process raw revision log data for HTML display.
Its purpose is to:
* add links to relevant browse views
* format date in human readable format
* truncate the message log
Args:
revision_log (list): raw revision log as returned by the swh-web api
per_page (int): number of log entries per page
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
"""
revision_log_data = []
for i, rev in enumerate(revision_log):
if i == per_page:
break
author_name = "None"
author_fullname = "None"
committer_fullname = "None"
if rev["author"]:
author_name = gen_person_mail_link(rev["author"])
author_fullname = rev["author"]["fullname"]
if rev["committer"]:
committer_fullname = rev["committer"]["fullname"]
author_date = format_utc_iso_date(rev["date"])
committer_date = format_utc_iso_date(rev["committer_date"])
tooltip = "revision %s\n" % rev["id"]
tooltip += "author: %s\n" % author_fullname
tooltip += "author date: %s\n" % author_date
tooltip += "committer: %s\n" % committer_fullname
tooltip += "committer date: %s\n\n" % committer_date
if rev["message"]:
tooltip += textwrap.indent(rev["message"], " " * 4)
revision_log_data.append(
{
"author": author_name,
"id": rev["id"][:7],
"message": rev["message"],
"date": author_date,
"commit_date": committer_date,
"url": gen_revision_url(rev["id"], snapshot_context),
"tooltip": tooltip,
}
)
return revision_log_data
# list of common readme names ordered by preference
# (lower indices have higher priority)
_common_readme_names = [
"readme.markdown",
"readme.md",
"readme.rst",
"readme.txt",
"readme",
]
def get_readme_to_display(readmes):
"""
Process a list of readme files found in a directory
in order to find the adequate one to display.
Args:
readmes: a list of dict where keys are readme file names and values
are readme sha1s
Returns:
A tuple (readme_name, readme_sha1)
"""
readme_name = None
readme_url = None
readme_sha1 = None
readme_html = None
lc_readmes = {k.lower(): {"orig_name": k, "sha1": v} for k, v in readmes.items()}
# look for readme names according to the preference order
# defined by the _common_readme_names list
for common_readme_name in _common_readme_names:
if common_readme_name in lc_readmes:
readme_name = lc_readmes[common_readme_name]["orig_name"]
readme_sha1 = lc_readmes[common_readme_name]["sha1"]
readme_url = reverse(
"browse-content-raw",
url_args={"query_string": readme_sha1},
query_params={"re_encode": "true"},
)
break
# otherwise pick the first readme like file if any
if not readme_name and len(readmes.items()) > 0:
readme_name = next(iter(readmes))
readme_sha1 = readmes[readme_name]
readme_url = reverse(
"browse-content-raw",
url_args={"query_string": readme_sha1},
query_params={"re_encode": "true"},
)
# convert rst README to html server side as there is
# no viable solution to perform that task client side
if readme_name and readme_name.endswith(".rst"):
cache_entry_id = "readme_%s" % readme_sha1
cache_entry = cache.get(cache_entry_id)
if cache_entry:
readme_html = cache_entry
else:
try:
rst_doc = request_content(readme_sha1)
readme_html = rst_to_html(rst_doc["raw_data"])
cache.set(cache_entry_id, readme_html)
except Exception as exc:
sentry_sdk.capture_exception(exc)
readme_html = "Readme bytes are not available"
return readme_name, readme_url, readme_html
diff --git a/swh/web/browse/views/content.py b/swh/web/browse/views/content.py
index 0bb83bd0..6975569c 100644
--- a/swh/web/browse/views/content.py
+++ b/swh/web/browse/views/content.py
@@ -1,407 +1,408 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import difflib
from distutils.util import strtobool
+import sentry_sdk
+
from django.http import HttpResponse, JsonResponse
from django.shortcuts import render
from django.template.defaultfilters import filesizeformat
-import sentry_sdk
from swh.model.hashutil import hash_to_hex
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.web.browse.browseurls import browse_route
from swh.web.browse.snapshot_context import get_snapshot_context
from swh.web.browse.utils import (
content_display_max_size,
gen_directory_link,
gen_link,
prepare_content_for_display,
request_content,
)
from swh.web.common import highlightjs, query, service
from swh.web.common.exc import NotFoundExc, handle_view_exception
from swh.web.common.identifiers import get_swhids_info
from swh.web.common.typing import ContentMetadata, SWHObjectInfo
from swh.web.common.utils import gen_path_info, reverse, swh_object_icons
@browse_route(
r"content/(?P<query_string>[0-9a-z_:]*[0-9a-f]+.)/raw/",
view_name="browse-content-raw",
checksum_args=["query_string"],
)
def content_raw(request, query_string):
"""Django view that produces a raw display of a content identified
by its hash value.
The url that points to it is
:http:get:`/browse/content/[(algo_hash):](hash)/raw/`
"""
try:
re_encode = bool(strtobool(request.GET.get("re_encode", "false")))
algo, checksum = query.parse_hash(query_string)
checksum = hash_to_hex(checksum)
content_data = request_content(query_string, max_size=None, re_encode=re_encode)
except Exception as exc:
return handle_view_exception(request, exc)
filename = request.GET.get("filename", None)
if not filename:
filename = "%s_%s" % (algo, checksum)
if (
content_data["mimetype"].startswith("text/")
or content_data["mimetype"] == "inode/x-empty"
):
response = HttpResponse(content_data["raw_data"], content_type="text/plain")
response["Content-disposition"] = "filename=%s" % filename
else:
response = HttpResponse(
content_data["raw_data"], content_type="application/octet-stream"
)
response["Content-disposition"] = "attachment; filename=%s" % filename
return response
_auto_diff_size_limit = 20000
@browse_route(
r"content/(?P<from_query_string>.*)/diff/(?P<to_query_string>.*)",
view_name="diff-contents",
)
def _contents_diff(request, from_query_string, to_query_string):
"""
Browse endpoint used to compute unified diffs between two contents.
Diffs are generated only if the two contents are textual.
By default, diffs whose size are greater than 20 kB will
not be generated. To force the generation of large diffs,
the 'force' boolean query parameter must be used.
Args:
request: input django http request
from_query_string: a string of the form "[ALGO_HASH:]HASH" where
optional ALGO_HASH can be either ``sha1``, ``sha1_git``,
``sha256``, or ``blake2s256`` (default to ``sha1``) and HASH
the hexadecimal representation of the hash value identifying
the first content
to_query_string: same as above for identifying the second content
Returns:
A JSON object containing the unified diff.
"""
diff_data = {}
content_from = None
content_to = None
content_from_size = 0
content_to_size = 0
content_from_lines = []
content_to_lines = []
force = request.GET.get("force", "false")
path = request.GET.get("path", None)
language = "nohighlight"
force = bool(strtobool(force))
if from_query_string == to_query_string:
diff_str = "File renamed without changes"
else:
try:
text_diff = True
if from_query_string:
content_from = request_content(from_query_string, max_size=None)
content_from_display_data = prepare_content_for_display(
content_from["raw_data"], content_from["mimetype"], path
)
language = content_from_display_data["language"]
content_from_size = content_from["length"]
if not (
content_from["mimetype"].startswith("text/")
or content_from["mimetype"] == "inode/x-empty"
):
text_diff = False
if text_diff and to_query_string:
content_to = request_content(to_query_string, max_size=None)
content_to_display_data = prepare_content_for_display(
content_to["raw_data"], content_to["mimetype"], path
)
language = content_to_display_data["language"]
content_to_size = content_to["length"]
if not (
content_to["mimetype"].startswith("text/")
or content_to["mimetype"] == "inode/x-empty"
):
text_diff = False
diff_size = abs(content_to_size - content_from_size)
if not text_diff:
diff_str = "Diffs are not generated for non textual content"
language = "nohighlight"
elif not force and diff_size > _auto_diff_size_limit:
diff_str = "Large diffs are not automatically computed"
language = "nohighlight"
else:
if content_from:
content_from_lines = (
content_from["raw_data"].decode("utf-8").splitlines(True)
)
if content_from_lines and content_from_lines[-1][-1] != "\n":
content_from_lines[-1] += "[swh-no-nl-marker]\n"
if content_to:
content_to_lines = (
content_to["raw_data"].decode("utf-8").splitlines(True)
)
if content_to_lines and content_to_lines[-1][-1] != "\n":
content_to_lines[-1] += "[swh-no-nl-marker]\n"
diff_lines = difflib.unified_diff(content_from_lines, content_to_lines)
diff_str = "".join(list(diff_lines)[2:])
except Exception as exc:
sentry_sdk.capture_exception(exc)
diff_str = str(exc)
diff_data["diff_str"] = diff_str
diff_data["language"] = language
return JsonResponse(diff_data)
@browse_route(
r"content/(?P<query_string>[0-9a-z_:]*[0-9a-f]+.)/",
view_name="browse-content",
checksum_args=["query_string"],
)
def content_display(request, query_string):
"""Django view that produces an HTML display of a content identified
by its hash value.
The url that points to it is
:http:get:`/browse/content/[(algo_hash):](hash)/`
"""
try:
algo, checksum = query.parse_hash(query_string)
checksum = hash_to_hex(checksum)
content_data = request_content(query_string, raise_if_unavailable=False)
origin_url = request.GET.get("origin_url")
selected_language = request.GET.get("language")
if not origin_url:
origin_url = request.GET.get("origin")
snapshot_id = request.GET.get("snapshot")
path = request.GET.get("path")
snapshot_context = None
if origin_url is not None or snapshot_id is not None:
try:
snapshot_context = get_snapshot_context(
origin_url=origin_url,
snapshot_id=snapshot_id,
branch_name=request.GET.get("branch"),
release_name=request.GET.get("release"),
revision_id=request.GET.get("revision"),
path=path,
browse_context=CONTENT,
)
except NotFoundExc as e:
if str(e).startswith("Origin"):
raw_cnt_url = reverse(
"browse-content", url_args={"query_string": query_string}
)
error_message = (
"The Software Heritage archive has a content "
"with the hash you provided but the origin "
"mentioned in your request appears broken: %s. "
"Please check the URL and try again.\n\n"
"Nevertheless, you can still browse the content "
"without origin information: %s"
% (gen_link(origin_url), gen_link(raw_cnt_url))
)
raise NotFoundExc(error_message)
else:
raise e
except Exception as exc:
return handle_view_exception(request, exc)
content = None
language = None
mimetype = None
if content_data["raw_data"] is not None:
content_display_data = prepare_content_for_display(
content_data["raw_data"], content_data["mimetype"], path
)
content = content_display_data["content_data"]
language = content_display_data["language"]
mimetype = content_display_data["mimetype"]
# Override language with user-selected language
if selected_language is not None:
language = selected_language
available_languages = None
if mimetype and "text/" in mimetype:
available_languages = highlightjs.get_supported_languages()
filename = None
path_info = None
directory_id = None
directory_url = None
root_dir = None
if snapshot_context:
root_dir = snapshot_context.get("root_directory")
query_params = snapshot_context["query_params"] if snapshot_context else {}
breadcrumbs = []
if path:
split_path = path.split("/")
root_dir = root_dir or split_path[0]
filename = split_path[-1]
if root_dir != path:
path = path.replace(root_dir + "/", "")
path = path[: -len(filename)]
path_info = gen_path_info(path)
query_params.pop("path", None)
dir_url = reverse(
"browse-directory",
url_args={"sha1_git": root_dir},
query_params=query_params,
)
breadcrumbs.append({"name": root_dir[:7], "url": dir_url})
for pi in path_info:
query_params["path"] = pi["path"]
dir_url = reverse(
"browse-directory",
url_args={"sha1_git": root_dir},
query_params=query_params,
)
breadcrumbs.append({"name": pi["name"], "url": dir_url})
breadcrumbs.append({"name": filename, "url": None})
if path and root_dir != path:
try:
dir_info = service.lookup_directory_with_path(root_dir, path)
directory_id = dir_info["target"]
except Exception as exc:
return handle_view_exception(request, exc)
elif root_dir != path:
directory_id = root_dir
else:
root_dir = None
if directory_id:
directory_url = gen_directory_link(directory_id)
query_params = {"filename": filename}
content_checksums = content_data["checksums"]
content_url = reverse(
"browse-content",
url_args={"query_string": f'sha1_git:{content_checksums["sha1_git"]}'},
)
content_raw_url = reverse(
"browse-content-raw",
url_args={"query_string": query_string},
query_params=query_params,
)
content_metadata = ContentMetadata(
object_type=CONTENT,
object_id=content_checksums["sha1_git"],
sha1=content_checksums["sha1"],
sha1_git=content_checksums["sha1_git"],
sha256=content_checksums["sha256"],
blake2s256=content_checksums["blake2s256"],
content_url=content_url,
mimetype=content_data["mimetype"],
encoding=content_data["encoding"],
size=filesizeformat(content_data["length"]),
language=content_data["language"],
licenses=content_data["licenses"],
root_directory=root_dir,
path=f"/{path}" if path else "",
filename=filename or "",
directory=directory_id,
directory_url=directory_url,
revision=None,
release=None,
snapshot=None,
origin_url=origin_url,
)
swh_objects = [
SWHObjectInfo(object_type=CONTENT, object_id=content_checksums["sha1_git"])
]
if directory_id:
swh_objects.append(SWHObjectInfo(object_type=DIRECTORY, object_id=directory_id))
if snapshot_context:
swh_objects.append(
SWHObjectInfo(
object_type=REVISION, object_id=snapshot_context["revision_id"]
)
)
swh_objects.append(
SWHObjectInfo(
object_type=SNAPSHOT, object_id=snapshot_context["snapshot_id"]
)
)
if snapshot_context["release_id"]:
swh_objects.append(
SWHObjectInfo(
object_type=RELEASE, object_id=snapshot_context["release_id"]
)
)
swhids_info = get_swhids_info(
swh_objects, snapshot_context, extra_context=content_metadata,
)
heading = "Content - %s" % content_checksums["sha1_git"]
if breadcrumbs:
content_path = "/".join([bc["name"] for bc in breadcrumbs])
heading += " - %s" % content_path
return render(
request,
"browse/content.html",
{
"heading": heading,
"swh_object_id": swhids_info[0]["swhid"],
"swh_object_name": "Content",
"swh_object_metadata": content_metadata,
"content": content,
"content_size": content_data["length"],
"max_content_size": content_display_max_size,
"filename": filename,
"encoding": content_data["encoding"],
"mimetype": mimetype,
"language": language,
"available_languages": available_languages,
"breadcrumbs": breadcrumbs,
"top_right_link": {
"url": content_raw_url,
"icon": swh_object_icons["content"],
"text": "Raw File",
},
"snapshot_context": snapshot_context,
"vault_cooking": None,
"show_actions": True,
"swhids_info": swhids_info,
"error_code": content_data["error_code"],
"error_message": content_data["error_message"],
"error_description": content_data["error_description"],
},
status=content_data["error_code"],
)
diff --git a/swh/web/browse/views/directory.py b/swh/web/browse/views/directory.py
index c4821de6..18c066c5 100644
--- a/swh/web/browse/views/directory.py
+++ b/swh/web/browse/views/directory.py
@@ -1,271 +1,272 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
+import sentry_sdk
+
from django.http import HttpResponse
from django.shortcuts import redirect, render
from django.template.defaultfilters import filesizeformat
-import sentry_sdk
from swh.model.identifiers import DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.web.browse.browseurls import browse_route
from swh.web.browse.snapshot_context import get_snapshot_context
from swh.web.browse.utils import gen_link, get_directory_entries, get_readme_to_display
from swh.web.common import service
from swh.web.common.exc import NotFoundExc, handle_view_exception
from swh.web.common.identifiers import get_swhids_info
from swh.web.common.typing import DirectoryMetadata, SWHObjectInfo
from swh.web.common.utils import gen_path_info, reverse, swh_object_icons
def _directory_browse(request, sha1_git, path=None):
root_sha1_git = sha1_git
try:
if path:
dir_info = service.lookup_directory_with_path(sha1_git, path)
sha1_git = dir_info["target"]
dirs, files = get_directory_entries(sha1_git)
origin_url = request.GET.get("origin_url")
if not origin_url:
origin_url = request.GET.get("origin")
snapshot_id = request.GET.get("snapshot")
snapshot_context = None
if origin_url is not None or snapshot_id is not None:
try:
snapshot_context = get_snapshot_context(
snapshot_id=snapshot_id,
origin_url=origin_url,
branch_name=request.GET.get("branch"),
release_name=request.GET.get("release"),
revision_id=request.GET.get("revision"),
path=path,
)
except NotFoundExc as e:
if str(e).startswith("Origin"):
raw_dir_url = reverse(
"browse-directory", url_args={"sha1_git": sha1_git}
)
error_message = (
"The Software Heritage archive has a directory "
"with the hash you provided but the origin "
"mentioned in your request appears broken: %s. "
"Please check the URL and try again.\n\n"
"Nevertheless, you can still browse the directory "
"without origin information: %s"
% (gen_link(origin_url), gen_link(raw_dir_url))
)
raise NotFoundExc(error_message)
else:
raise e
except Exception as exc:
return handle_view_exception(request, exc)
path_info = gen_path_info(path)
query_params = snapshot_context["query_params"] if snapshot_context else {}
breadcrumbs = []
breadcrumbs.append(
{
"name": root_sha1_git[:7],
"url": reverse(
"browse-directory",
url_args={"sha1_git": root_sha1_git},
query_params=query_params,
),
}
)
for pi in path_info:
breadcrumbs.append(
{
"name": pi["name"],
"url": reverse(
"browse-directory",
url_args={"sha1_git": root_sha1_git},
query_params={"path": pi["path"], **query_params},
),
}
)
path = "" if path is None else (path + "/")
for d in dirs:
if d["type"] == "rev":
d["url"] = reverse(
"browse-revision",
url_args={"sha1_git": d["target"]},
query_params=query_params,
)
else:
d["url"] = reverse(
"browse-directory",
url_args={"sha1_git": root_sha1_git},
query_params={"path": path + d["name"], **query_params},
)
sum_file_sizes = 0
readmes = {}
for f in files:
query_string = "sha1_git:" + f["target"]
f["url"] = reverse(
"browse-content",
url_args={"query_string": query_string},
query_params={
"path": root_sha1_git + "/" + path + f["name"],
**query_params,
},
)
if f["length"] is not None:
sum_file_sizes += f["length"]
f["length"] = filesizeformat(f["length"])
if f["name"].lower().startswith("readme"):
readmes[f["name"]] = f["checksums"]["sha1"]
readme_name, readme_url, readme_html = get_readme_to_display(readmes)
sum_file_sizes = filesizeformat(sum_file_sizes)
dir_metadata = DirectoryMetadata(
object_type=DIRECTORY,
object_id=sha1_git,
directory=root_sha1_git,
nb_files=len(files),
nb_dirs=len(dirs),
sum_file_sizes=sum_file_sizes,
root_directory=root_sha1_git,
path=f"/{path}" if path else "/",
revision=None,
revision_found=None,
release=None,
snapshot=None,
)
vault_cooking = {
"directory_context": True,
"directory_id": sha1_git,
"revision_context": False,
"revision_id": None,
}
swh_objects = [SWHObjectInfo(object_type=DIRECTORY, object_id=sha1_git)]
if snapshot_context:
swh_objects.append(
SWHObjectInfo(
object_type=REVISION, object_id=snapshot_context["revision_id"]
)
)
swh_objects.append(
SWHObjectInfo(
object_type=SNAPSHOT, object_id=snapshot_context["snapshot_id"]
)
)
if snapshot_context["release_id"]:
swh_objects.append(
SWHObjectInfo(
object_type=RELEASE, object_id=snapshot_context["release_id"]
)
)
swhids_info = get_swhids_info(swh_objects, snapshot_context, dir_metadata)
heading = "Directory - %s" % sha1_git
if breadcrumbs:
dir_path = "/".join([bc["name"] for bc in breadcrumbs]) + "/"
heading += " - %s" % dir_path
top_right_link = None
if snapshot_context is not None and not snapshot_context["is_empty"]:
history_url = reverse(
"browse-revision-log",
url_args={"sha1_git": snapshot_context["revision_id"]},
query_params=query_params,
)
top_right_link = {
"url": history_url,
"icon": swh_object_icons["revisions history"],
"text": "History",
}
return render(
request,
"browse/directory.html",
{
"heading": heading,
"swh_object_id": swhids_info[0]["swhid"],
"swh_object_name": "Directory",
"swh_object_metadata": dir_metadata,
"dirs": dirs,
"files": files,
"breadcrumbs": breadcrumbs,
"top_right_link": top_right_link,
"readme_name": readme_name,
"readme_url": readme_url,
"readme_html": readme_html,
"snapshot_context": snapshot_context,
"vault_cooking": vault_cooking,
"show_actions": True,
"swhids_info": swhids_info,
},
)
@browse_route(
r"directory/(?P<sha1_git>[0-9a-f]+)/",
view_name="browse-directory",
checksum_args=["sha1_git"],
)
def directory_browse(request, sha1_git):
"""Django view for browsing the content of a directory identified
by its sha1_git value.
The url that points to it is
:http:get:`/browse/directory/(sha1_git)/`
"""
return _directory_browse(request, sha1_git, request.GET.get("path"))
@browse_route(
r"directory/(?P<sha1_git>[0-9a-f]+)/(?P<path>.+)/",
view_name="browse-directory-legacy",
checksum_args=["sha1_git"],
)
def directory_browse_legacy(request, sha1_git, path):
"""Django view for browsing the content of a directory identified
by its sha1_git value.
The url that points to it is
:http:get:`/browse/directory/(sha1_git)/(path)/`
"""
return _directory_browse(request, sha1_git, path)
@browse_route(
r"directory/resolve/content-path/(?P<sha1_git>[0-9a-f]+)/",
view_name="browse-directory-resolve-content-path",
checksum_args=["sha1_git"],
)
def _directory_resolve_content_path(request, sha1_git):
"""
Internal endpoint redirecting to data url for a specific file path
relative to a root directory.
"""
try:
path = os.path.normpath(request.GET.get("path"))
if not path.startswith("../"):
dir_info = service.lookup_directory_with_path(sha1_git, path)
if dir_info["type"] == "file":
sha1 = dir_info["checksums"]["sha1"]
data_url = reverse(
"browse-content-raw", url_args={"query_string": sha1}
)
return redirect(data_url)
except Exception as exc:
sentry_sdk.capture_exception(exc)
return HttpResponse(status=404)
diff --git a/swh/web/browse/views/release.py b/swh/web/browse/views/release.py
index d95de1b7..c3162865 100644
--- a/swh/web/browse/views/release.py
+++ b/swh/web/browse/views/release.py
@@ -1,241 +1,242 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
-from django.shortcuts import render
import sentry_sdk
+from django.shortcuts import render
+
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.web.browse.browseurls import browse_route
from swh.web.browse.snapshot_context import get_snapshot_context
from swh.web.browse.utils import (
gen_content_link,
gen_directory_link,
gen_link,
gen_person_mail_link,
gen_release_link,
gen_revision_link,
gen_snapshot_link,
)
from swh.web.common import service
from swh.web.common.exc import NotFoundExc, handle_view_exception
from swh.web.common.identifiers import get_swhids_info
from swh.web.common.typing import ReleaseMetadata, SWHObjectInfo
from swh.web.common.utils import format_utc_iso_date, reverse
@browse_route(
r"release/(?P<sha1_git>[0-9a-f]+)/",
view_name="browse-release",
checksum_args=["sha1_git"],
)
def release_browse(request, sha1_git):
"""
Django view that produces an HTML display of a release
identified by its id.
The url that points to it is :http:get:`/browse/release/(sha1_git)/`.
"""
try:
release = service.lookup_release(sha1_git)
snapshot_context = {}
origin_info = None
snapshot_id = request.GET.get("snapshot_id")
if not snapshot_id:
snapshot_id = request.GET.get("snapshot")
origin_url = request.GET.get("origin_url")
if not origin_url:
origin_url = request.GET.get("origin")
timestamp = request.GET.get("timestamp")
visit_id = request.GET.get("visit_id")
if origin_url:
try:
snapshot_context = get_snapshot_context(
snapshot_id, origin_url, timestamp, visit_id
)
except NotFoundExc as e:
raw_rel_url = reverse("browse-release", url_args={"sha1_git": sha1_git})
error_message = (
"The Software Heritage archive has a release "
"with the hash you provided but the origin "
"mentioned in your request appears broken: %s. "
"Please check the URL and try again.\n\n"
"Nevertheless, you can still browse the release "
"without origin information: %s"
% (gen_link(origin_url), gen_link(raw_rel_url))
)
if str(e).startswith("Origin"):
raise NotFoundExc(error_message)
else:
raise e
origin_info = snapshot_context["origin_info"]
elif snapshot_id:
snapshot_context = get_snapshot_context(snapshot_id)
except Exception as exc:
return handle_view_exception(request, exc)
target_url = None
if release["target_type"] == REVISION:
target_url = gen_revision_link(release["target"])
elif release["target_type"] == CONTENT:
target_url = gen_content_link(release["target"])
elif release["target_type"] == DIRECTORY:
target_url = gen_directory_link(release["target"])
elif release["target_type"] == RELEASE:
target_url = gen_release_link(release["target"])
snapshot_id = None
browse_snp_link = None
if snapshot_context:
snapshot_id = snapshot_context["snapshot_id"]
browse_snp_link = gen_snapshot_link(snapshot_id)
release_metadata = ReleaseMetadata(
object_type=RELEASE,
object_id=sha1_git,
release=sha1_git,
release_url=gen_release_link(release["id"]),
author=release["author"]["fullname"] if release["author"] else "None",
author_url=gen_person_mail_link(release["author"])
if release["author"]
else "None",
date=format_utc_iso_date(release["date"]),
name=release["name"],
synthetic=release["synthetic"],
target=release["target"],
target_type=release["target_type"],
target_url=target_url,
snapshot=snapshot_context.get("snapshot_id", None),
snapshot_url=browse_snp_link,
origin_url=origin_url,
)
release_note_lines = []
if release["message"]:
release_note_lines = release["message"].split("\n")
vault_cooking = None
rev_directory = None
target_link = None
if release["target_type"] == REVISION:
target_link = gen_revision_link(
release["target"],
snapshot_context=snapshot_context,
link_text=None,
link_attrs=None,
)
try:
revision = service.lookup_revision(release["target"])
rev_directory = revision["directory"]
vault_cooking = {
"directory_context": True,
"directory_id": rev_directory,
"revision_context": True,
"revision_id": release["target"],
}
except Exception as exc:
sentry_sdk.capture_exception(exc)
elif release["target_type"] == DIRECTORY:
target_link = gen_directory_link(
release["target"],
snapshot_context=snapshot_context,
link_text=None,
link_attrs=None,
)
try:
# check directory exists
service.lookup_directory(release["target"])
vault_cooking = {
"directory_context": True,
"directory_id": release["target"],
"revision_context": False,
"revision_id": None,
}
except Exception as exc:
sentry_sdk.capture_exception(exc)
elif release["target_type"] == CONTENT:
target_link = gen_content_link(
release["target"],
snapshot_context=snapshot_context,
link_text=None,
link_attrs=None,
)
elif release["target_type"] == RELEASE:
target_link = gen_release_link(
release["target"],
snapshot_context=snapshot_context,
link_text=None,
link_attrs=None,
)
rev_directory_url = None
if rev_directory is not None:
if origin_info:
rev_directory_url = reverse(
"browse-origin-directory",
query_params={
"origin_url": origin_info["url"],
"release": release["name"],
"snapshot": snapshot_id,
},
)
elif snapshot_id:
rev_directory_url = reverse(
"browse-snapshot-directory",
url_args={"snapshot_id": snapshot_id},
query_params={"release": release["name"]},
)
else:
rev_directory_url = reverse(
"browse-directory", url_args={"sha1_git": rev_directory}
)
directory_link = None
if rev_directory_url is not None:
directory_link = gen_link(rev_directory_url, rev_directory)
release["directory_link"] = directory_link
release["target_link"] = target_link
swh_objects = [SWHObjectInfo(object_type=RELEASE, object_id=sha1_git)]
if snapshot_context:
snapshot_id = snapshot_context["snapshot_id"]
if snapshot_id:
swh_objects.append(SWHObjectInfo(object_type=SNAPSHOT, object_id=snapshot_id))
swhids_info = get_swhids_info(swh_objects, snapshot_context)
note_header = "None"
if len(release_note_lines) > 0:
note_header = release_note_lines[0]
release["note_header"] = note_header
release["note_body"] = "\n".join(release_note_lines[1:])
heading = "Release - %s" % release["name"]
if snapshot_context:
context_found = "snapshot: %s" % snapshot_context["snapshot_id"]
if origin_info:
context_found = "origin: %s" % origin_info["url"]
heading += " - %s" % context_found
return render(
request,
"browse/release.html",
{
"heading": heading,
"swh_object_id": swhids_info[0]["swhid"],
"swh_object_name": "Release",
"swh_object_metadata": release_metadata,
"release": release,
"snapshot_context": snapshot_context,
"show_actions": True,
"breadcrumbs": None,
"vault_cooking": vault_cooking,
"top_right_link": None,
"swhids_info": swhids_info,
},
)
diff --git a/swh/web/common/exc.py b/swh/web/common/exc.py
index 2e18a5b3..f3c9550c 100644
--- a/swh/web/common/exc.py
+++ b/swh/web/common/exc.py
@@ -1,149 +1,150 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import traceback
+import sentry_sdk
+
from django.http import HttpResponse
from django.shortcuts import render
from django.utils.html import escape
from django.utils.safestring import mark_safe
-import sentry_sdk
from swh.web.config import get_config
class BadInputExc(ValueError):
"""Wrong request to the api.
Example: Asking a content with the wrong identifier format.
"""
pass
class NotFoundExc(Exception):
"""Good request to the api but no result were found.
Example: Asking a content with the right identifier format but
that content does not exist.
"""
pass
class ForbiddenExc(Exception):
"""Good request to the api, forbidden result to return due to enforce
policy.
Example: Asking for a raw content which exists but whose mimetype
is not text.
"""
pass
class LargePayloadExc(Exception):
"""The input size is too large.
Example: Asking to resolve 10000 SWHIDs when the limit is 1000.
"""
pass
http_status_code_message = {
400: "Bad Request",
401: "Unauthorized",
403: "Access Denied",
404: "Resource not found",
413: "Payload Too Large",
500: "Internal Server Error",
501: "Not Implemented",
502: "Bad Gateway",
503: "Service unavailable",
}
def _generate_error_page(request, error_code, error_description):
return render(
request,
"error.html",
{
"error_code": error_code,
"error_message": http_status_code_message[error_code],
"error_description": mark_safe(error_description),
},
status=error_code,
)
def swh_handle400(request, exception=None):
"""
Custom Django HTTP error 400 handler for swh-web.
"""
error_description = (
"The server cannot process the request to %s due to "
"something that is perceived to be a client error."
% escape(request.META["PATH_INFO"])
)
return _generate_error_page(request, 400, error_description)
def swh_handle403(request, exception=None):
"""
Custom Django HTTP error 403 handler for swh-web.
"""
error_description = "The resource %s requires an authentication." % escape(
request.META["PATH_INFO"]
)
return _generate_error_page(request, 403, error_description)
def swh_handle404(request, exception=None):
"""
Custom Django HTTP error 404 handler for swh-web.
"""
error_description = "The resource %s could not be found on the server." % escape(
request.META["PATH_INFO"]
)
return _generate_error_page(request, 404, error_description)
def swh_handle500(request):
"""
Custom Django HTTP error 500 handler for swh-web.
"""
error_description = (
"An unexpected condition was encountered when "
"requesting resource %s." % escape(request.META["PATH_INFO"])
)
return _generate_error_page(request, 500, error_description)
def handle_view_exception(request, exc, html_response=True):
"""
Function used to generate an error page when an exception
was raised inside a swh-web browse view.
"""
sentry_sdk.capture_exception(exc)
error_code = 500
error_description = "%s: %s" % (type(exc).__name__, str(exc))
if get_config()["debug"]:
error_description = traceback.format_exc()
if isinstance(exc, BadInputExc):
error_code = 400
if isinstance(exc, ForbiddenExc):
error_code = 403
if isinstance(exc, NotFoundExc):
error_code = 404
if html_response:
return _generate_error_page(request, error_code, error_description)
else:
return HttpResponse(
error_description, content_type="text/plain", status=error_code
)
diff --git a/swh/web/common/identifiers.py b/swh/web/common/identifiers.py
index 48bdb6b4..0f9b0f12 100644
--- a/swh/web/common/identifiers.py
+++ b/swh/web/common/identifiers.py
@@ -1,382 +1,383 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from typing import Any, Dict, Iterable, List, Optional, cast
from urllib.parse import quote
-from django.http import QueryDict
from typing_extensions import TypedDict
+from django.http import QueryDict
+
from swh.model.exceptions import ValidationError
from swh.model.hashutil import hash_to_bytes
from swh.model.identifiers import (
CONTENT,
DIRECTORY,
ORIGIN,
RELEASE,
REVISION,
SNAPSHOT,
SWHID,
parse_swhid,
swhid,
)
from swh.web.common import service
from swh.web.common.exc import BadInputExc
from swh.web.common.typing import (
QueryParameters,
SnapshotContext,
SWHIDContext,
SWHIDInfo,
SWHObjectInfo,
)
from swh.web.common.utils import reverse
def gen_swhid(
object_type: str,
object_id: str,
scheme_version: int = 1,
metadata: SWHIDContext = {},
) -> str:
"""
Returns the SoftWare Heritage persistent IDentifier for a swh object based on:
* the object type
* the object id
* the SWHID scheme version
Args:
object_type: the swh object type
(content/directory/release/revision/snapshot)
object_id: the swh object id (hexadecimal representation
of its hash value)
scheme_version: the scheme version of the SWHIDs
Returns:
the SWHID of the object
Raises:
BadInputExc: if the provided parameters do not enable to
generate a valid identifier
"""
try:
obj_swhid = swhid(
object_type, object_id, scheme_version, cast(Dict[str, Any], metadata)
)
except ValidationError as e:
raise BadInputExc("Invalid object (%s) for SWHID. %s" % (object_id, e))
else:
return obj_swhid
class ResolvedSWHID(TypedDict):
"""parsed SWHID with context"""
swhid_parsed: SWHID
"""URL to browse object according to SWHID context"""
browse_url: Optional[str]
def resolve_swhid(
swhid: str, query_params: Optional[QueryParameters] = None
) -> ResolvedSWHID:
"""
Try to resolve a SoftWare Heritage persistent IDentifier into an url for
browsing the targeted object.
Args:
swhid: a SoftWare Heritage persistent IDentifier
query_params: optional dict filled with
query parameters to append to the browse url
Returns:
a dict with the following keys:
* **swhid_parsed**: the parsed identifier
* **browse_url**: the url for browsing the targeted object
"""
swhid_parsed = get_swhid(swhid)
object_type = swhid_parsed.object_type
object_id = swhid_parsed.object_id
browse_url = None
url_args = {}
query_dict = QueryDict("", mutable=True)
fragment = ""
anchor_swhid_parsed = None
process_lines = object_type is CONTENT
if query_params and len(query_params) > 0:
for k in sorted(query_params.keys()):
query_dict[k] = query_params[k]
if "origin" in swhid_parsed.metadata:
query_dict["origin_url"] = swhid_parsed.metadata["origin"]
if "anchor" in swhid_parsed.metadata:
anchor_swhid_parsed = get_swhid(swhid_parsed.metadata["anchor"])
if "path" in swhid_parsed.metadata and swhid_parsed.metadata["path"] != "/":
query_dict["path"] = swhid_parsed.metadata["path"]
if anchor_swhid_parsed:
directory = ""
if anchor_swhid_parsed.object_type == DIRECTORY:
directory = anchor_swhid_parsed.object_id
elif anchor_swhid_parsed.object_type == REVISION:
revision = service.lookup_revision(anchor_swhid_parsed.object_id)
directory = revision["directory"]
elif anchor_swhid_parsed.object_type == RELEASE:
release = service.lookup_release(anchor_swhid_parsed.object_id)
if release["target_type"] == REVISION:
revision = service.lookup_revision(release["target"])
directory = revision["directory"]
if object_type == CONTENT:
if "origin" not in swhid_parsed.metadata:
# when no origin context, content objects need to have their
# path prefixed by root directory id for proper breadcrumbs display
query_dict["path"] = directory + query_dict["path"]
else:
# remove leading slash from SWHID content path
query_dict["path"] = query_dict["path"][1:]
elif object_type == DIRECTORY:
object_id = directory
# remove leading and trailing slashes from SWHID directory path
query_dict["path"] = query_dict["path"][1:-1]
# snapshot context
if "visit" in swhid_parsed.metadata:
snp_swhid_parsed = get_swhid(swhid_parsed.metadata["visit"])
if snp_swhid_parsed.object_type != SNAPSHOT:
raise BadInputExc("Visit must be a snapshot SWHID.")
query_dict["snapshot"] = snp_swhid_parsed.object_id
if anchor_swhid_parsed:
if anchor_swhid_parsed.object_type == REVISION:
# check if the anchor revision is the tip of a branch
branch_name = service.lookup_snapshot_branch_name_from_tip_revision(
snp_swhid_parsed.object_id, anchor_swhid_parsed.object_id
)
if branch_name:
query_dict["branch"] = branch_name
elif object_type != REVISION:
query_dict["revision"] = anchor_swhid_parsed.object_id
elif anchor_swhid_parsed.object_type == RELEASE:
release = service.lookup_release(anchor_swhid_parsed.object_id)
if release:
query_dict["release"] = release["name"]
if object_type == REVISION and "release" not in query_dict:
branch_name = service.lookup_snapshot_branch_name_from_tip_revision(
snp_swhid_parsed.object_id, object_id
)
if branch_name:
query_dict["branch"] = branch_name
# browsing content or directory without snapshot context
elif object_type in (CONTENT, DIRECTORY) and anchor_swhid_parsed:
if anchor_swhid_parsed.object_type == REVISION:
# anchor revision, objects are browsed from its view
object_type = REVISION
object_id = anchor_swhid_parsed.object_id
elif object_type == DIRECTORY and anchor_swhid_parsed.object_type == DIRECTORY:
# a directory is browsed from its root
object_id = anchor_swhid_parsed.object_id
if object_type == CONTENT:
url_args["query_string"] = f"sha1_git:{object_id}"
elif object_type == DIRECTORY:
url_args["sha1_git"] = object_id
elif object_type == RELEASE:
url_args["sha1_git"] = object_id
elif object_type == REVISION:
url_args["sha1_git"] = object_id
elif object_type == SNAPSHOT:
url_args["snapshot_id"] = object_id
elif object_type == ORIGIN:
raise BadInputExc(
(
"Origin SWHIDs are not publicly resolvable because they are for "
"internal usage only"
)
)
if "lines" in swhid_parsed.metadata and process_lines:
lines = swhid_parsed.metadata["lines"].split("-")
fragment += "#L" + lines[0]
if len(lines) > 1:
fragment += "-L" + lines[1]
if url_args:
browse_url = (
reverse(
f"browse-{object_type}", url_args=url_args, query_params=query_dict,
)
+ fragment
)
return ResolvedSWHID(swhid_parsed=swhid_parsed, browse_url=browse_url)
def get_swhid(swhid: str) -> SWHID:
"""Check if a SWHID is valid and return it parsed.
Args:
swhid: a SoftWare Heritage persistent IDentifier.
Raises:
BadInputExc: if the provided SWHID can not be parsed.
Return:
A parsed SWHID.
"""
try:
swhid_parsed = parse_swhid(swhid)
except ValidationError as ve:
raise BadInputExc("Error when parsing identifier: %s" % " ".join(ve.messages))
else:
return swhid_parsed
def group_swhids(swhids: Iterable[SWHID],) -> Dict[str, List[bytes]]:
"""
Groups many SoftWare Heritage persistent IDentifiers into a
dictionary depending on their type.
Args:
swhids: an iterable of SoftWare Heritage persistent
IDentifier objects
Returns:
A dictionary with:
keys: object types
values: object hashes
"""
swhids_by_type: Dict[str, List[bytes]] = {
CONTENT: [],
DIRECTORY: [],
REVISION: [],
RELEASE: [],
SNAPSHOT: [],
}
for obj_swhid in swhids:
obj_id = obj_swhid.object_id
obj_type = obj_swhid.object_type
swhids_by_type[obj_type].append(hash_to_bytes(obj_id))
return swhids_by_type
def get_swhids_info(
swh_objects: Iterable[SWHObjectInfo],
snapshot_context: Optional[SnapshotContext] = None,
extra_context: Optional[Dict[str, Any]] = None,
) -> List[SWHIDInfo]:
"""
Returns a list of dict containing info related to SWHIDs of objects.
Args:
swh_objects: an iterable of dict describing archived objects
snapshot_context: optional dict parameter describing the snapshot in
which the objects have been found
extra_context: optional dict filled with extra contextual info about
the objects
Returns:
a list of dict containing SWHIDs info
"""
swhids_info = []
for swh_object in swh_objects:
if not swh_object["object_id"]:
swhids_info.append(
SWHIDInfo(
object_type=swh_object["object_type"],
object_id="",
swhid="",
swhid_url="",
context={},
swhid_with_context=None,
swhid_with_context_url=None,
)
)
continue
object_type = swh_object["object_type"]
object_id = swh_object["object_id"]
swhid_context: SWHIDContext = {}
if snapshot_context:
if snapshot_context["origin_info"] is not None:
swhid_context["origin"] = quote(
snapshot_context["origin_info"]["url"], safe="/?:@&"
)
if object_type != SNAPSHOT:
swhid_context["visit"] = gen_swhid(
SNAPSHOT, snapshot_context["snapshot_id"]
)
if object_type in (CONTENT, DIRECTORY):
if snapshot_context["release_id"] is not None:
swhid_context["anchor"] = gen_swhid(
RELEASE, snapshot_context["release_id"]
)
elif snapshot_context["revision_id"] is not None:
swhid_context["anchor"] = gen_swhid(
REVISION, snapshot_context["revision_id"]
)
if object_type in (CONTENT, DIRECTORY):
if (
extra_context
and "revision" in extra_context
and extra_context["revision"]
and "anchor" not in swhid_context
):
swhid_context["anchor"] = gen_swhid(REVISION, extra_context["revision"])
elif (
extra_context
and "root_directory" in extra_context
and extra_context["root_directory"]
and "anchor" not in swhid_context
and (
object_type != DIRECTORY
or extra_context["root_directory"] != object_id
)
):
swhid_context["anchor"] = gen_swhid(
DIRECTORY, extra_context["root_directory"]
)
path = None
if extra_context and "path" in extra_context:
path = extra_context["path"] or "/"
if "filename" in extra_context and object_type == CONTENT:
path += extra_context["filename"]
if path:
swhid_context["path"] = quote(path, safe="/?:@&")
swhid = gen_swhid(object_type, object_id)
swhid_url = reverse("browse-swhid", url_args={"swhid": swhid})
swhid_with_context = None
swhid_with_context_url = None
if swhid_context:
swhid_with_context = gen_swhid(
object_type, object_id, metadata=swhid_context
)
swhid_with_context_url = reverse(
"browse-swhid", url_args={"swhid": swhid_with_context}
)
swhids_info.append(
SWHIDInfo(
object_type=object_type,
object_id=object_id,
swhid=swhid,
swhid_url=swhid_url,
context=swhid_context,
swhid_with_context=swhid_with_context,
swhid_with_context_url=swhid_with_context_url,
)
)
return swhids_info
diff --git a/swh/web/common/origin_save.py b/swh/web/common/origin_save.py
index 47426e3d..510f1797 100644
--- a/swh/web/common/origin_save.py
+++ b/swh/web/common/origin_save.py
@@ -1,616 +1,617 @@
# Copyright (C) 2018-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from bisect import bisect_right
from datetime import datetime, timedelta, timezone
from itertools import product
import json
import logging
from typing import Any, Dict
-from django.core.exceptions import ObjectDoesNotExist, ValidationError
-from django.core.validators import URLValidator
-from django.utils.html import escape
from prometheus_client import Gauge
import requests
import sentry_sdk
+from django.core.exceptions import ObjectDoesNotExist, ValidationError
+from django.core.validators import URLValidator
+from django.utils.html import escape
+
from swh.scheduler.utils import create_oneshot_task_dict
from swh.web import config
from swh.web.common import service
from swh.web.common.exc import BadInputExc, ForbiddenExc, NotFoundExc
from swh.web.common.models import (
SAVE_REQUEST_ACCEPTED,
SAVE_REQUEST_PENDING,
SAVE_REQUEST_REJECTED,
SAVE_TASK_FAILED,
SAVE_TASK_NOT_CREATED,
SAVE_TASK_NOT_YET_SCHEDULED,
SAVE_TASK_RUNNING,
SAVE_TASK_SCHEDULED,
SAVE_TASK_SUCCEED,
SaveAuthorizedOrigin,
SaveOriginRequest,
SaveUnauthorizedOrigin,
)
from swh.web.common.origin_visits import get_origin_visits
from swh.web.common.utils import SWH_WEB_METRICS_REGISTRY, parse_iso8601_date_to_utc
scheduler = config.scheduler()
logger = logging.getLogger(__name__)
def get_origin_save_authorized_urls():
"""
Get the list of origin url prefixes authorized to be
immediately loaded into the archive (whitelist).
Returns:
list: The list of authorized origin url prefix
"""
return [origin.url for origin in SaveAuthorizedOrigin.objects.all()]
def get_origin_save_unauthorized_urls():
"""
Get the list of origin url prefixes forbidden to be
loaded into the archive (blacklist).
Returns:
list: the list of unauthorized origin url prefix
"""
return [origin.url for origin in SaveUnauthorizedOrigin.objects.all()]
def can_save_origin(origin_url):
"""
Check if a software origin can be saved into the archive.
Based on the origin url, the save request will be either:
* immediately accepted if the url is whitelisted
* rejected if the url is blacklisted
* put in pending state for manual review otherwise
Args:
origin_url (str): the software origin url to check
Returns:
str: the origin save request status, either **accepted**,
**rejected** or **pending**
"""
# origin url may be blacklisted
for url_prefix in get_origin_save_unauthorized_urls():
if origin_url.startswith(url_prefix):
return SAVE_REQUEST_REJECTED
# if the origin url is in the white list, it can be immediately saved
for url_prefix in get_origin_save_authorized_urls():
if origin_url.startswith(url_prefix):
return SAVE_REQUEST_ACCEPTED
# otherwise, the origin url needs to be manually verified
return SAVE_REQUEST_PENDING
# map visit type to scheduler task
# TODO: do not hardcode the task name here (T1157)
_visit_type_task = {"git": "load-git", "hg": "load-hg", "svn": "load-svn"}
# map scheduler task status to origin save status
_save_task_status = {
"next_run_not_scheduled": SAVE_TASK_NOT_YET_SCHEDULED,
"next_run_scheduled": SAVE_TASK_SCHEDULED,
"completed": SAVE_TASK_SUCCEED,
"disabled": SAVE_TASK_FAILED,
}
def get_savable_visit_types():
return sorted(list(_visit_type_task.keys()))
def _check_visit_type_savable(visit_type):
"""
Get the list of visit types that can be performed
through a save request.
Returns:
list: the list of saveable visit types
"""
allowed_visit_types = ", ".join(get_savable_visit_types())
if visit_type not in _visit_type_task:
raise BadInputExc(
"Visit of type %s can not be saved! "
"Allowed types are the following: %s" % (visit_type, allowed_visit_types)
)
_validate_url = URLValidator(schemes=["http", "https", "svn", "git"])
def _check_origin_url_valid(origin_url):
try:
_validate_url(origin_url)
except ValidationError:
raise BadInputExc(
"The provided origin url (%s) is not valid!" % escape(origin_url)
)
def _get_visit_info_for_save_request(save_request):
visit_date = None
visit_status = None
time_now = datetime.now(tz=timezone.utc)
time_delta = time_now - save_request.request_date
# stop trying to find a visit date one month after save request submission
# as those requests to storage are expensive and associated loading task
# surely ended up with errors
if time_delta.days <= 30:
try:
origin = {"url": save_request.origin_url}
origin_info = service.lookup_origin(origin)
origin_visits = get_origin_visits(origin_info)
visit_dates = [parse_iso8601_date_to_utc(v["date"]) for v in origin_visits]
i = bisect_right(visit_dates, save_request.request_date)
if i != len(visit_dates):
visit_date = visit_dates[i]
visit_status = origin_visits[i]["status"]
if origin_visits[i]["status"] not in ("full", "partial"):
visit_date = None
except Exception as exc:
sentry_sdk.capture_exception(exc)
return visit_date, visit_status
def _check_visit_update_status(save_request, save_task_status):
visit_date, visit_status = _get_visit_info_for_save_request(save_request)
save_request.visit_date = visit_date
# visit has been performed, mark the saving task as succeed
if visit_date and visit_status is not None:
save_task_status = SAVE_TASK_SUCCEED
elif visit_status == "ongoing":
save_task_status = SAVE_TASK_RUNNING
else:
time_now = datetime.now(tz=timezone.utc)
time_delta = time_now - save_request.request_date
# consider the task as failed if it is still in scheduled state
# 30 days after its submission
if time_delta.days > 30:
save_task_status = SAVE_TASK_FAILED
return visit_date, save_task_status
def _save_request_dict(save_request, task=None):
must_save = False
visit_date = save_request.visit_date
# save task still in scheduler db
if task:
save_task_status = _save_task_status[task["status"]]
# Consider request from which a visit date has already been found
# as succeeded to avoid retrieving it again
if save_task_status == SAVE_TASK_SCHEDULED and visit_date:
save_task_status = SAVE_TASK_SUCCEED
if save_task_status in (SAVE_TASK_FAILED, SAVE_TASK_SUCCEED) and not visit_date:
visit_date, _ = _get_visit_info_for_save_request(save_request)
save_request.visit_date = visit_date
must_save = True
# Check tasks still marked as scheduled / not yet scheduled
if save_task_status in (SAVE_TASK_SCHEDULED, SAVE_TASK_NOT_YET_SCHEDULED):
visit_date, save_task_status = _check_visit_update_status(
save_request, save_task_status
)
# save task may have been archived
else:
save_task_status = save_request.loading_task_status
if save_task_status in (SAVE_TASK_SCHEDULED, SAVE_TASK_NOT_YET_SCHEDULED):
visit_date, save_task_status = _check_visit_update_status(
save_request, save_task_status
)
else:
save_task_status = save_request.loading_task_status
if save_request.loading_task_status != save_task_status:
save_request.loading_task_status = save_task_status
must_save = True
if must_save:
save_request.save()
return {
"id": save_request.id,
"visit_type": save_request.visit_type,
"origin_url": save_request.origin_url,
"save_request_date": save_request.request_date.isoformat(),
"save_request_status": save_request.status,
"save_task_status": save_task_status,
"visit_date": visit_date.isoformat() if visit_date else None,
}
def create_save_origin_request(visit_type, origin_url):
"""
Create a loading task to save a software origin into the archive.
This function aims to create a software origin loading task
trough the use of the swh-scheduler component.
First, some checks are performed to see if the visit type and origin
url are valid but also if the the save request can be accepted.
If those checks passed, the loading task is then created.
Otherwise, the save request is put in pending or rejected state.
All the submitted save requests are logged into the swh-web
database to keep track of them.
Args:
visit_type (str): the type of visit to perform (currently only
``git`` but ``svn`` and ``hg`` will soon be available)
origin_url (str): the url of the origin to save
Raises:
BadInputExc: the visit type or origin url is invalid
ForbiddenExc: the provided origin url is blacklisted
Returns:
dict: A dict describing the save request with the following keys:
* **visit_type**: the type of visit to perform
* **origin_url**: the url of the origin
* **save_request_date**: the date the request was submitted
* **save_request_status**: the request status, either **accepted**,
**rejected** or **pending**
* **save_task_status**: the origin loading task status, either
**not created**, **not yet scheduled**, **scheduled**,
**succeed** or **failed**
"""
_check_visit_type_savable(visit_type)
_check_origin_url_valid(origin_url)
save_request_status = can_save_origin(origin_url)
task = None
# if the origin save request is accepted, create a scheduler
# task to load it into the archive
if save_request_status == SAVE_REQUEST_ACCEPTED:
# create a task with high priority
kwargs = {
"priority": "high",
"url": origin_url,
}
sor = None
# get list of previously sumitted save requests
current_sors = list(
SaveOriginRequest.objects.filter(
visit_type=visit_type, origin_url=origin_url
)
)
can_create_task = False
# if no save requests previously submitted, create the scheduler task
if not current_sors:
can_create_task = True
else:
# get the latest submitted save request
sor = current_sors[0]
# if it was in pending state, we need to create the scheduler task
# and update the save request info in the database
if sor.status == SAVE_REQUEST_PENDING:
can_create_task = True
# a task has already been created to load the origin
elif sor.loading_task_id != -1:
# get the scheduler task and its status
tasks = scheduler.get_tasks([sor.loading_task_id])
task = tasks[0] if tasks else None
task_status = _save_request_dict(sor, task)["save_task_status"]
# create a new scheduler task only if the previous one has been
# already executed
if task_status == SAVE_TASK_FAILED or task_status == SAVE_TASK_SUCCEED:
can_create_task = True
sor = None
else:
can_create_task = False
if can_create_task:
# effectively create the scheduler task
task_dict = create_oneshot_task_dict(_visit_type_task[visit_type], **kwargs)
task = scheduler.create_tasks([task_dict])[0]
# pending save request has been accepted
if sor:
sor.status = SAVE_REQUEST_ACCEPTED
sor.loading_task_id = task["id"]
sor.save()
else:
sor = SaveOriginRequest.objects.create(
visit_type=visit_type,
origin_url=origin_url,
status=save_request_status,
loading_task_id=task["id"],
)
# save request must be manually reviewed for acceptation
elif save_request_status == SAVE_REQUEST_PENDING:
# check if there is already such a save request already submitted,
# no need to add it to the database in that case
try:
sor = SaveOriginRequest.objects.get(
visit_type=visit_type, origin_url=origin_url, status=save_request_status
)
# if not add it to the database
except ObjectDoesNotExist:
sor = SaveOriginRequest.objects.create(
visit_type=visit_type, origin_url=origin_url, status=save_request_status
)
# origin can not be saved as its url is blacklisted,
# log the request to the database anyway
else:
sor = SaveOriginRequest.objects.create(
visit_type=visit_type, origin_url=origin_url, status=save_request_status
)
if save_request_status == SAVE_REQUEST_REJECTED:
raise ForbiddenExc(
(
'The "save code now" request has been rejected '
"because the provided origin url is blacklisted."
)
)
return _save_request_dict(sor, task)
def get_save_origin_requests_from_queryset(requests_queryset):
"""
Get all save requests from a SaveOriginRequest queryset.
Args:
requests_queryset (django.db.models.QuerySet): input
SaveOriginRequest queryset
Returns:
list: A list of save origin requests dict as described in
:func:`swh.web.common.origin_save.create_save_origin_request`
"""
task_ids = []
for sor in requests_queryset:
task_ids.append(sor.loading_task_id)
save_requests = []
if task_ids:
tasks = scheduler.get_tasks(task_ids)
tasks = {task["id"]: task for task in tasks}
for sor in requests_queryset:
sr_dict = _save_request_dict(sor, tasks.get(sor.loading_task_id))
save_requests.append(sr_dict)
return save_requests
def get_save_origin_requests(visit_type, origin_url):
"""
Get all save requests for a given software origin.
Args:
visit_type (str): the type of visit
origin_url (str): the url of the origin
Raises:
BadInputExc: the visit type or origin url is invalid
swh.web.common.exc.NotFoundExc: no save requests can be found for the
given origin
Returns:
list: A list of save origin requests dict as described in
:func:`swh.web.common.origin_save.create_save_origin_request`
"""
_check_visit_type_savable(visit_type)
_check_origin_url_valid(origin_url)
sors = SaveOriginRequest.objects.filter(
visit_type=visit_type, origin_url=origin_url
)
if sors.count() == 0:
raise NotFoundExc(
("No save requests found for visit of type " "%s on origin with url %s.")
% (visit_type, origin_url)
)
return get_save_origin_requests_from_queryset(sors)
def get_save_origin_task_info(
save_request_id: int, full_info: bool = True
) -> Dict[str, Any]:
"""
Get detailed information about an accepted save origin request
and its associated loading task.
If the associated loading task info is archived and removed
from the scheduler database, returns an empty dictionary.
Args:
save_request_id: identifier of a save origin request
full_info: whether to return detailed info for staff users
Returns:
A dictionary with the following keys:
- **type**: loading task type
- **arguments**: loading task arguments
- **id**: loading task database identifier
- **backend_id**: loading task celery identifier
- **scheduled**: loading task scheduling date
- **ended**: loading task termination date
- **status**: loading task execution status
Depending on the availability of the task logs in the elasticsearch
cluster of Software Heritage, the returned dictionary may also
contain the following keys:
- **name**: associated celery task name
- **message**: relevant log message from task execution
- **duration**: task execution time (only if it succeeded)
- **worker**: name of the worker that executed the task
"""
try:
save_request = SaveOriginRequest.objects.get(id=save_request_id)
except ObjectDoesNotExist:
return {}
task = scheduler.get_tasks([save_request.loading_task_id])
task = task[0] if task else None
if task is None:
return {}
task_run = scheduler.get_task_runs([task["id"]])
task_run = task_run[0] if task_run else None
if task_run is None:
return {}
task_run["type"] = task["type"]
task_run["arguments"] = task["arguments"]
task_run["id"] = task_run["task"]
del task_run["task"]
del task_run["metadata"]
es_workers_index_url = config.get_config()["es_workers_index_url"]
if not es_workers_index_url:
return task_run
es_workers_index_url += "/_search"
if save_request.visit_date:
min_ts = save_request.visit_date
max_ts = min_ts + timedelta(days=7)
else:
min_ts = save_request.request_date
max_ts = min_ts + timedelta(days=30)
min_ts_unix = int(min_ts.timestamp()) * 1000
max_ts_unix = int(max_ts.timestamp()) * 1000
save_task_status = _save_task_status[task["status"]]
priority = "3" if save_task_status == SAVE_TASK_FAILED else "6"
query = {
"bool": {
"must": [
{"match_phrase": {"priority": {"query": priority}}},
{"match_phrase": {"swh_task_id": {"query": task_run["backend_id"]}}},
{
"range": {
"@timestamp": {
"gte": min_ts_unix,
"lte": max_ts_unix,
"format": "epoch_millis",
}
}
},
]
}
}
try:
response = requests.post(
es_workers_index_url,
json={"query": query, "sort": ["@timestamp"]},
timeout=30,
)
results = json.loads(response.text)
if results["hits"]["total"]["value"] >= 1:
task_run_info = results["hits"]["hits"][-1]["_source"]
if "swh_logging_args_runtime" in task_run_info:
duration = task_run_info["swh_logging_args_runtime"]
task_run["duration"] = duration
if "message" in task_run_info:
task_run["message"] = task_run_info["message"]
if "swh_logging_args_name" in task_run_info:
task_run["name"] = task_run_info["swh_logging_args_name"]
elif "swh_task_name" in task_run_info:
task_run["name"] = task_run_info["swh_task_name"]
if "hostname" in task_run_info:
task_run["worker"] = task_run_info["hostname"]
elif "host" in task_run_info:
task_run["worker"] = task_run_info["host"]
except Exception as exc:
logger.warning("Request to Elasticsearch failed\n%s", exc)
sentry_sdk.capture_exception(exc)
if not full_info:
for field in ("id", "backend_id", "worker"):
# remove some staff only fields
task_run.pop(field, None)
if "message" in task_run and "Loading failure" in task_run["message"]:
# hide traceback for non staff users, only display exception
message_lines = task_run["message"].split("\n")
message = ""
for line in message_lines:
if line.startswith("Traceback"):
break
message += f"{line}\n"
message += message_lines[-1]
task_run["message"] = message
return task_run
SUBMITTED_SAVE_REQUESTS_METRIC = "swh_web_submitted_save_requests"
_submitted_save_requests_gauge = Gauge(
name=SUBMITTED_SAVE_REQUESTS_METRIC,
documentation="Number of submitted origin save requests",
labelnames=["status", "visit_type"],
registry=SWH_WEB_METRICS_REGISTRY,
)
ACCEPTED_SAVE_REQUESTS_METRIC = "swh_web_accepted_save_requests"
_accepted_save_requests_gauge = Gauge(
name=ACCEPTED_SAVE_REQUESTS_METRIC,
documentation="Number of accepted origin save requests",
labelnames=["load_task_status", "visit_type"],
registry=SWH_WEB_METRICS_REGISTRY,
)
def compute_save_requests_metrics():
"""Compute a couple of Prometheus metrics related to
origin save requests"""
request_statuses = (
SAVE_REQUEST_ACCEPTED,
SAVE_REQUEST_REJECTED,
SAVE_REQUEST_PENDING,
)
load_task_statuses = (
SAVE_TASK_NOT_CREATED,
SAVE_TASK_NOT_YET_SCHEDULED,
SAVE_TASK_SCHEDULED,
SAVE_TASK_SUCCEED,
SAVE_TASK_FAILED,
SAVE_TASK_RUNNING,
)
visit_types = get_savable_visit_types()
labels_set = product(request_statuses, visit_types)
for labels in labels_set:
_submitted_save_requests_gauge.labels(*labels).set(0)
labels_set = product(load_task_statuses, visit_types)
for labels in labels_set:
_accepted_save_requests_gauge.labels(*labels).set(0)
for sor in SaveOriginRequest.objects.all():
if sor.status == SAVE_REQUEST_ACCEPTED:
_accepted_save_requests_gauge.labels(
load_task_status=sor.loading_task_status, visit_type=sor.visit_type
).inc()
_submitted_save_requests_gauge.labels(
status=sor.status, visit_type=sor.visit_type
).inc()
diff --git a/swh/web/common/swh_templatetags.py b/swh/web/common/swh_templatetags.py
index 9245c9e1..174dc1b8 100644
--- a/swh/web/common/swh_templatetags.py
+++ b/swh/web/common/swh_templatetags.py
@@ -1,156 +1,157 @@
# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import json
import re
+import sentry_sdk
+
from django import template
from django.core.serializers.json import DjangoJSONEncoder
from django.utils.safestring import mark_safe
-import sentry_sdk
from swh.web.common.origin_save import get_savable_visit_types
from swh.web.common.utils import rst_to_html
register = template.Library()
@register.filter
def docstring_display(docstring):
"""
Utility function to htmlize reST-formatted documentation in browsable
api.
"""
return rst_to_html(docstring)
@register.filter
def urlize_links_and_mails(text):
"""Utility function for decorating api links in browsable api.
Args:
text: whose content matching links should be transformed into
contextual API or Browse html links.
Returns
The text transformed if any link is found.
The text as is otherwise.
"""
try:
if 'href="' not in text:
text = re.sub(r"(http.*)", r'<a href="\1">\1</a>', text)
return re.sub(r'([^ <>"]+@[^ <>"]+)', r'<a href="mailto:\1">\1</a>', text)
except Exception as exc:
sentry_sdk.capture_exception(exc)
return text
@register.filter
def urlize_header_links(text):
"""Utility function for decorating headers links in browsable api.
Args
text: Text whose content contains Link header value
Returns:
The text transformed with html link if any link is found.
The text as is otherwise.
"""
links = text.split(",")
ret = ""
for i, link in enumerate(links):
ret += re.sub(r"<(http.*)>", r'<<a href="\1">\1</a>>', link)
# add one link per line and align them
if i != len(links) - 1:
ret += "\n "
return ret
@register.filter
def jsonify(obj):
"""Utility function for converting a django template variable
to JSON in order to use it in script tags.
Args
obj: Any django template context variable
Returns:
JSON representation of the variable.
"""
return mark_safe(json.dumps(obj, cls=DjangoJSONEncoder))
@register.filter
def sub(value, arg):
"""Django template filter for subtracting two numbers
Args:
value (int/float): the value to subtract from
arg (int/float): the value to subtract to
Returns:
int/float: The subtraction result
"""
return value - arg
@register.filter
def mul(value, arg):
"""Django template filter for multiplying two numbers
Args:
value (int/float): the value to multiply from
arg (int/float): the value to multiply with
Returns:
int/float: The multiplication result
"""
return value * arg
@register.filter
def key_value(dict, key):
"""Django template filter to get a value in a dictionary.
Args:
dict (dict): a dictionary
key (str): the key to lookup value
Returns:
The requested value in the dictionary
"""
return dict[key]
@register.filter
def visit_type_savable(visit_type):
"""Django template filter to check if a save request can be
created for a given visit type.
Args:
visit_type (str): the type of visit
Returns:
If the visit type is saveable or not
"""
return visit_type in get_savable_visit_types()
@register.filter
def split(value, arg):
"""Django template filter to split a string.
Args:
value (str): the string to split
arg (str): the split separator
Returns:
list: the split string parts
"""
return value.split(arg)
diff --git a/swh/web/common/typing.py b/swh/web/common/typing.py
index 953562ed..6a027ac2 100644
--- a/swh/web/common/typing.py
+++ b/swh/web/common/typing.py
@@ -1,223 +1,224 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from typing import Any, Dict, List, Optional, TypeVar, Union
-from django.http import QueryDict
from typing_extensions import TypedDict
+from django.http import QueryDict
+
from swh.core.api.classes import PagedResult as CorePagedResult
QueryParameters = Union[Dict[str, Any], QueryDict]
class OriginInfo(TypedDict):
url: str
"""URL of the origin"""
class OriginMetadataInfo(TypedDict):
url: str
"""URL of the origin"""
metadata: Dict[str, Any]
"""Origin metadata associated to the origin"""
class OriginVisitInfo(TypedDict):
date: str
"""date of the visit in iso format"""
formatted_date: str
"""formatted date of the visit"""
metadata: Dict[str, Any]
"""metadata associated to the visit"""
origin: str
"""visited origin URL"""
snapshot: str
"""snapshot identifier computed during the visit"""
status: str
"""status of the visit ("ongoing", "full" or "partial") """
type: str
"""visit type (git, hg, debian, ...)"""
url: str
"""URL to browse the snapshot"""
visit: int
"""visit identifier"""
class SnapshotBranchInfo(TypedDict):
date: Optional[str]
""""author date of branch heading revision"""
directory: Optional[str]
"""directory associated to branch heading revision"""
message: Optional[str]
"""message of branch heading revision"""
name: Optional[str]
"""branch name"""
revision: str
"""branch heading revision"""
url: Optional[str]
"""optional browse URL (content, directory, ...) scoped to branch"""
class SnapshotReleaseInfo(TypedDict):
branch_name: str
"""branch name associated to release in snapshot"""
date: str
"""release date"""
directory: Optional[str]
"""optional directory associatd to the release"""
id: str
"""release identifier"""
message: str
"""release message"""
name: str
"""release name"""
target: str
"""release target"""
target_type: str
"""release target_type"""
url: Optional[str]
"""optional browse URL (content, directory, ...) scoped to release"""
class SnapshotContext(TypedDict):
branch: Optional[str]
"""optional branch name set when browsing snapshot in that scope"""
branches: List[SnapshotBranchInfo]
"""list of snapshot branches (possibly truncated)"""
branches_url: str
"""snapshot branches list browse URL"""
is_empty: bool
"""indicates if the snapshot is empty"""
origin_info: Optional[OriginInfo]
"""optional origin info associated to the snapshot"""
origin_visits_url: Optional[str]
"""optional origin visits URL"""
query_params: QueryParameters
"""common query parameters when browsing snapshot content"""
release: Optional[str]
"""optional release name set when browsing snapshot in that scope"""
release_id: Optional[str]
"""optional release identifier set when browsing snapshot in that scope"""
releases: List[SnapshotReleaseInfo]
"""list of snapshot releases (possibly truncated)"""
releases_url: str
"""snapshot releases list browse URL"""
revision_id: Optional[str]
"""optional revision identifier set when browsing snapshot in that scope"""
revision_info: Optional[Dict[str, Any]]
"""optional revision info set when browsing snapshot in that scope"""
root_directory: Optional[str]
"""optional root directory identifier set when browsing snapshot content"""
snapshot_id: str
"""snapshot identifier"""
snapshot_sizes: Dict[str, int]
"""snapshot sizes grouped by branch target type"""
snapshot_swhid: str
"""snapshot SWHID"""
url_args: Dict[str, Any]
"""common URL arguments when browsing snapshot content"""
visit_info: Optional[OriginVisitInfo]
"""optional origin visit info associated to the snapshot"""
class SWHObjectInfo(TypedDict):
object_type: str
object_id: str
class SWHIDContext(TypedDict, total=False):
origin: str
anchor: str
visit: str
path: str
lines: str
class SWHIDInfo(SWHObjectInfo):
swhid: str
swhid_url: str
context: SWHIDContext
swhid_with_context: Optional[str]
swhid_with_context_url: Optional[str]
class SWHObjectInfoMetadata(TypedDict, total=False):
origin_url: Optional[str]
visit_date: Optional[str]
visit_type: Optional[str]
directory_url: Optional[str]
revision_url: Optional[str]
release_url: Optional[str]
snapshot_url: Optional[str]
class ContentMetadata(SWHObjectInfo, SWHObjectInfoMetadata):
sha1: str
sha1_git: str
sha256: str
blake2s256: str
content_url: str
mimetype: str
encoding: str
size: str
language: str
licenses: str
path: Optional[str]
filename: Optional[str]
directory: Optional[str]
root_directory: Optional[str]
revision: Optional[str]
release: Optional[str]
snapshot: Optional[str]
class DirectoryMetadata(SWHObjectInfo, SWHObjectInfoMetadata):
directory: str
nb_files: int
nb_dirs: int
sum_file_sizes: str
root_directory: Optional[str]
path: str
revision: Optional[str]
revision_found: Optional[bool]
release: Optional[str]
snapshot: Optional[str]
class ReleaseMetadata(SWHObjectInfo, SWHObjectInfoMetadata):
release: str
author: str
author_url: str
date: str
name: str
synthetic: bool
target: str
target_type: str
target_url: str
snapshot: Optional[str]
class RevisionMetadata(SWHObjectInfo, SWHObjectInfoMetadata):
revision: str
author: str
author_url: str
committer: str
committer_url: str
date: str
committer_date: str
directory: str
merge: bool
metadata: str
parents: List[str]
synthetic: bool
type: str
snapshot: Optional[str]
TResult = TypeVar("TResult")
PagedResult = CorePagedResult[TResult, str]
diff --git a/swh/web/common/utils.py b/swh/web/common/utils.py
index 5fe3316b..27931f95 100644
--- a/swh/web/common/utils.py
+++ b/swh/web/common/utils.py
@@ -1,349 +1,350 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime, timezone
import re
from typing import Any, Dict, Optional
from bs4 import BeautifulSoup
-from django.http import HttpRequest, QueryDict
-from django.urls import reverse as django_reverse
from docutils.core import publish_parts
import docutils.parsers.rst
import docutils.utils
from docutils.writers.html5_polyglot import HTMLTranslator, Writer
from iso8601 import ParseError, parse_date
from prometheus_client.registry import CollectorRegistry
+
+from django.http import HttpRequest, QueryDict
+from django.urls import reverse as django_reverse
from rest_framework.authentication import SessionAuthentication
from swh.web.common.exc import BadInputExc
from swh.web.common.typing import QueryParameters
from swh.web.config import get_config
SWH_WEB_METRICS_REGISTRY = CollectorRegistry(auto_describe=True)
swh_object_icons = {
"branch": "mdi mdi-source-branch",
"branches": "mdi mdi-source-branch",
"content": "mdi mdi-file-document",
"directory": "mdi mdi-folder",
"origin": "mdi mdi-source-repository",
"person": "mdi mdi-account",
"revisions history": "mdi mdi-history",
"release": "mdi mdi-tag",
"releases": "mdi mdi-tag",
"revision": "mdi mdi-rotate-90 mdi-source-commit",
"snapshot": "mdi mdi-camera",
"visits": "mdi mdi-calendar-month",
}
def reverse(
viewname: str,
url_args: Optional[Dict[str, Any]] = None,
query_params: Optional[QueryParameters] = None,
current_app: Optional[str] = None,
urlconf: Optional[str] = None,
request: Optional[HttpRequest] = None,
) -> str:
"""An override of django reverse function supporting query parameters.
Args:
viewname: the name of the django view from which to compute a url
url_args: dictionary of url arguments indexed by their names
query_params: dictionary of query parameters to append to the
reversed url
current_app: the name of the django app tighten to the view
urlconf: url configuration module
request: build an absolute URI if provided
Returns:
str: the url of the requested view with processed arguments and
query parameters
"""
if url_args:
url_args = {k: v for k, v in url_args.items() if v is not None}
url = django_reverse(
viewname, urlconf=urlconf, kwargs=url_args, current_app=current_app
)
if query_params:
query_params = {k: v for k, v in query_params.items() if v is not None}
if query_params and len(query_params) > 0:
query_dict = QueryDict("", mutable=True)
for k in sorted(query_params.keys()):
query_dict[k] = query_params[k]
url += "?" + query_dict.urlencode(safe="/;:")
if request is not None:
url = request.build_absolute_uri(url)
return url
def datetime_to_utc(date):
"""Returns datetime in UTC without timezone info
Args:
date (datetime.datetime): input datetime with timezone info
Returns:
datetime.datetime: datetime in UTC without timezone info
"""
if date.tzinfo and date.tzinfo != timezone.utc:
return date.astimezone(tz=timezone.utc)
else:
return date
def parse_iso8601_date_to_utc(iso_date: str) -> datetime:
"""Given an ISO 8601 datetime string, parse the result as UTC datetime.
Returns:
a timezone-aware datetime representing the parsed date
Raises:
swh.web.common.exc.BadInputExc: provided date does not respect ISO 8601 format
Samples:
- 2016-01-12
- 2016-01-12T09:19:12+0100
- 2007-01-14T20:34:22Z
"""
try:
date = parse_date(iso_date)
return datetime_to_utc(date)
except ParseError as e:
raise BadInputExc(e)
def shorten_path(path):
"""Shorten the given path: for each hash present, only return the first
8 characters followed by an ellipsis"""
sha256_re = r"([0-9a-f]{8})[0-9a-z]{56}"
sha1_re = r"([0-9a-f]{8})[0-9a-f]{32}"
ret = re.sub(sha256_re, r"\1...", path)
return re.sub(sha1_re, r"\1...", ret)
def format_utc_iso_date(iso_date, fmt="%d %B %Y, %H:%M UTC"):
"""Turns a string representation of an ISO 8601 datetime string
to UTC and format it into a more human readable one.
For instance, from the following input
string: '2017-05-04T13:27:13+02:00' the following one
is returned: '04 May 2017, 11:27 UTC'.
Custom format string may also be provided
as parameter
Args:
iso_date (str): a string representation of an ISO 8601 date
fmt (str): optional date formatting string
Returns:
str: a formatted string representation of the input iso date
"""
if not iso_date:
return iso_date
date = parse_iso8601_date_to_utc(iso_date)
return date.strftime(fmt)
def gen_path_info(path):
"""Function to generate path data navigation for use
with a breadcrumb in the swh web ui.
For instance, from a path /folder1/folder2/folder3,
it returns the following list::
[{'name': 'folder1', 'path': 'folder1'},
{'name': 'folder2', 'path': 'folder1/folder2'},
{'name': 'folder3', 'path': 'folder1/folder2/folder3'}]
Args:
path: a filesystem path
Returns:
list: a list of path data for navigation as illustrated above.
"""
path_info = []
if path:
sub_paths = path.strip("/").split("/")
path_from_root = ""
for p in sub_paths:
path_from_root += "/" + p
path_info.append({"name": p, "path": path_from_root.strip("/")})
return path_info
def parse_rst(text, report_level=2):
"""
Parse a reStructuredText string with docutils.
Args:
text (str): string with reStructuredText markups in it
report_level (int): level of docutils report messages to print
(1 info 2 warning 3 error 4 severe 5 none)
Returns:
docutils.nodes.document: a parsed docutils document
"""
parser = docutils.parsers.rst.Parser()
components = (docutils.parsers.rst.Parser,)
settings = docutils.frontend.OptionParser(
components=components
).get_default_values()
settings.report_level = report_level
document = docutils.utils.new_document("rst-doc", settings=settings)
parser.parse(text, document)
return document
def get_client_ip(request):
"""
Return the client IP address from an incoming HTTP request.
Args:
request (django.http.HttpRequest): the incoming HTTP request
Returns:
str: The client IP address
"""
x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
if x_forwarded_for:
ip = x_forwarded_for.split(",")[0]
else:
ip = request.META.get("REMOTE_ADDR")
return ip
browsers_supported_image_mimes = set(
[
"image/gif",
"image/png",
"image/jpeg",
"image/bmp",
"image/webp",
"image/svg",
"image/svg+xml",
]
)
def context_processor(request):
"""
Django context processor used to inject variables
in all swh-web templates.
"""
config = get_config()
if (
hasattr(request, "user")
and request.user.is_authenticated
and not hasattr(request.user, "backend")
):
# To avoid django.template.base.VariableDoesNotExist errors
# when rendering templates when standard Django user is logged in.
request.user.backend = "django.contrib.auth.backends.ModelBackend"
return {
"swh_object_icons": swh_object_icons,
"available_languages": None,
"swh_client_config": config["client_config"],
"oidc_enabled": bool(config["keycloak"]["server_url"]),
"browsers_supported_image_mimes": browsers_supported_image_mimes,
}
class EnforceCSRFAuthentication(SessionAuthentication):
"""
Helper class to enforce CSRF validation on a DRF view
when a user is not authenticated.
"""
def authenticate(self, request):
user = getattr(request._request, "user", None)
self.enforce_csrf(request)
return (user, None)
def resolve_branch_alias(
snapshot: Dict[str, Any], branch: Optional[Dict[str, Any]]
) -> Optional[Dict[str, Any]]:
"""
Resolve branch alias in snapshot content.
Args:
snapshot: a full snapshot content
branch: a branch alias contained in the snapshot
Returns:
The real snapshot branch that got aliased.
"""
while branch and branch["target_type"] == "alias":
if branch["target"] in snapshot["branches"]:
branch = snapshot["branches"][branch["target"]]
else:
from swh.web.common import service
snp = service.lookup_snapshot(
snapshot["id"], branches_from=branch["target"], branches_count=1
)
if snp and branch["target"] in snp["branches"]:
branch = snp["branches"][branch["target"]]
else:
branch = None
return branch
class _NoHeaderHTMLTranslator(HTMLTranslator):
"""
Docutils translator subclass to customize the generation of HTML
from reST-formatted docstrings
"""
def __init__(self, document):
super().__init__(document)
self.body_prefix = []
self.body_suffix = []
_HTML_WRITER = Writer()
_HTML_WRITER.translator_class = _NoHeaderHTMLTranslator
def rst_to_html(rst: str) -> str:
"""
Convert reStructuredText document into HTML.
Args:
rst: A string containing a reStructuredText document
Returns:
Body content of the produced HTML conversion.
"""
settings = {
"initial_header_level": 2,
}
pp = publish_parts(rst, writer=_HTML_WRITER, settings_overrides=settings)
return f'<div class="swh-rst">{pp["html_body"]}</div>'
def prettify_html(html: str) -> str:
"""
Prettify an HTML document.
Args:
html: Input HTML document
Returns:
The prettified HTML document
"""
return BeautifulSoup(html, "lxml").prettify()
diff --git a/swh/web/misc/badges.py b/swh/web/misc/badges.py
index 37b45a3d..6281c2c8 100644
--- a/swh/web/misc/badges.py
+++ b/swh/web/misc/badges.py
@@ -1,165 +1,166 @@
# Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from base64 import b64encode
from typing import Optional, cast
+from pybadges import badge
+
from django.conf.urls import url
from django.contrib.staticfiles import finders
from django.http import HttpRequest, HttpResponse
-from pybadges import badge
from swh.model.exceptions import ValidationError
from swh.model.identifiers import (
CONTENT,
DIRECTORY,
ORIGIN,
RELEASE,
REVISION,
SNAPSHOT,
parse_swhid,
swhid,
)
from swh.web.common import service
from swh.web.common.exc import BadInputExc, NotFoundExc
from swh.web.common.identifiers import resolve_swhid
from swh.web.common.utils import reverse
_orange = "#f36a24"
_blue = "#0172b2"
_red = "#cd5741"
_swh_logo_data = None
_badge_config = {
CONTENT: {"color": _blue, "title": "Archived source file",},
DIRECTORY: {"color": _blue, "title": "Archived source tree",},
ORIGIN: {"color": _orange, "title": "Archived software repository",},
RELEASE: {"color": _blue, "title": "Archived software release",},
REVISION: {"color": _blue, "title": "Archived commit",},
SNAPSHOT: {"color": _blue, "title": "Archived software repository snapshot",},
"error": {"color": _red, "title": "An error occurred when generating the badge"},
}
def _get_logo_data() -> str:
"""
Get data-URI for Software Heritage SVG logo to embed it in
the generated badges.
"""
global _swh_logo_data
if _swh_logo_data is None:
swh_logo_path = cast(str, finders.find("img/swh-logo-white.svg"))
with open(swh_logo_path, "rb") as swh_logo_file:
_swh_logo_data = "data:image/svg+xml;base64,%s" % b64encode(
swh_logo_file.read()
).decode("ascii")
return _swh_logo_data
def _swh_badge(
request: HttpRequest,
object_type: str,
object_id: str,
object_swhid: Optional[str] = "",
) -> HttpResponse:
"""
Generate a Software Heritage badge for a given object type and id.
Args:
request: input http request
object_type: The type of swh object to generate a badge for,
either *content*, *directory*, *revision*, *release*, *origin*
or *snapshot*
object_id: The id of the swh object, either an url for origin
type or a *sha1* for other object types
object_swhid: If provided, the object SWHID will not be recomputed
Returns:
HTTP response with content type *image/svg+xml* containing the SVG
badge data. If the provided parameters are invalid, HTTP 400 status
code will be returned. If the object can not be found in the archive,
HTTP 404 status code will be returned.
"""
left_text = "error"
whole_link = None
try:
if object_type == ORIGIN:
service.lookup_origin({"url": object_id})
right_text = "repository"
whole_link = reverse(
"browse-origin", query_params={"origin_url": object_id}
)
else:
# when SWHID is provided, object type and id will be parsed
# from it
if object_swhid:
parsed_swhid = parse_swhid(object_swhid)
object_type = parsed_swhid.object_type
object_id = parsed_swhid.object_id
swh_object = service.lookup_object(object_type, object_id)
if object_swhid:
right_text = object_swhid
else:
right_text = swhid(object_type, object_id)
whole_link = resolve_swhid(right_text)["browse_url"]
# remove SWHID metadata if any for badge text
if object_swhid:
right_text = right_text.split(";")[0]
# use release name for badge text
if object_type == RELEASE:
right_text = "release %s" % swh_object["name"]
left_text = "archived"
except (BadInputExc, ValidationError):
right_text = f'invalid {object_type if object_type else "object"} id'
object_type = "error"
except NotFoundExc:
right_text = f'{object_type if object_type else "object"} not found'
object_type = "error"
badge_data = badge(
left_text=left_text,
right_text=right_text,
right_color=_badge_config[object_type]["color"],
whole_link=request.build_absolute_uri(whole_link),
whole_title=_badge_config[object_type]["title"],
logo=_get_logo_data(),
embed_logo=True,
)
return HttpResponse(badge_data, content_type="image/svg+xml")
def _swh_badge_swhid(request: HttpRequest, object_swhid: str) -> HttpResponse:
"""
Generate a Software Heritage badge for a given object SWHID.
Args:
request (django.http.HttpRequest): input http request
object_swhid (str): a SWHID of an archived object
Returns:
django.http.HttpResponse: An http response with content type
*image/svg+xml* containing the SVG badge data. If any error
occurs, a status code of 400 will be returned.
"""
return _swh_badge(request, "", "", object_swhid)
urlpatterns = [
url(
r"^badge/(?P<object_type>[a-z]+)/(?P<object_id>.+)/$",
_swh_badge,
name="swh-badge",
),
url(
r"^badge/(?P<object_swhid>swh:[0-9]+:[a-z]+:[0-9a-f]+.*)/$",
_swh_badge_swhid,
name="swh-badge-swhid",
),
]
diff --git a/swh/web/misc/metrics.py b/swh/web/misc/metrics.py
index f35d51cf..8ee7d39d 100644
--- a/swh/web/misc/metrics.py
+++ b/swh/web/misc/metrics.py
@@ -1,20 +1,21 @@
# Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
-from django.http import HttpResponse
from prometheus_client.exposition import CONTENT_TYPE_LATEST, generate_latest
+from django.http import HttpResponse
+
from swh.web.common.origin_save import compute_save_requests_metrics
from swh.web.common.utils import SWH_WEB_METRICS_REGISTRY
def prometheus_metrics(request):
compute_save_requests_metrics()
return HttpResponse(
content=generate_latest(registry=SWH_WEB_METRICS_REGISTRY),
content_type=CONTENT_TYPE_LATEST,
)
diff --git a/swh/web/misc/urls.py b/swh/web/misc/urls.py
index d4f835a4..cc00273c 100644
--- a/swh/web/misc/urls.py
+++ b/swh/web/misc/urls.py
@@ -1,101 +1,102 @@
# Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import json
+import requests
+import sentry_sdk
+
from django.conf.urls import include, url
from django.contrib.staticfiles import finders
from django.http import JsonResponse
from django.shortcuts import render
-import requests
-import sentry_sdk
from swh.web.common import service
from swh.web.config import get_config
from swh.web.misc.metrics import prometheus_metrics
def _jslicenses(request):
jslicenses_file = finders.find("jssources/jslicenses.json")
jslicenses_data = json.load(open(jslicenses_file))
jslicenses_data = sorted(
jslicenses_data.items(), key=lambda item: item[0].split("/")[-1]
)
return render(request, "misc/jslicenses.html", {"jslicenses_data": jslicenses_data})
def _stat_counters(request):
stat_counters = service.stat_counters()
url = get_config()["history_counters_url"]
stat_counters_history = "null"
if url:
try:
response = requests.get(url, timeout=5)
stat_counters_history = json.loads(response.text)
except Exception as exc:
sentry_sdk.capture_exception(exc)
counters = {
"stat_counters": stat_counters,
"stat_counters_history": stat_counters_history,
}
return JsonResponse(counters)
urlpatterns = [
url(r"^", include("swh.web.misc.coverage")),
url(r"^jslicenses/$", _jslicenses, name="jslicenses"),
url(r"^", include("swh.web.misc.origin_save")),
url(r"^stat_counters/", _stat_counters, name="stat-counters"),
url(r"^", include("swh.web.misc.badges")),
url(r"^metrics/prometheus/$", prometheus_metrics, name="metrics-prometheus"),
]
# when running end to end tests trough cypress, declare some extra
# endpoints to provide input data for some of those tests
if get_config()["e2e_tests_mode"]:
from swh.web.tests.views import (
get_content_code_data_all_exts,
get_content_code_data_all_filenames,
get_content_code_data_by_ext,
get_content_code_data_by_filename,
get_content_other_data_by_ext,
)
urlpatterns.append(
url(
r"^tests/data/content/code/extension/(?P<ext>.+)/$",
get_content_code_data_by_ext,
name="tests-content-code-extension",
)
)
urlpatterns.append(
url(
r"^tests/data/content/other/extension/(?P<ext>.+)/$",
get_content_other_data_by_ext,
name="tests-content-other-extension",
)
)
urlpatterns.append(
url(
r"^tests/data/content/code/extensions/$",
get_content_code_data_all_exts,
name="tests-content-code-extensions",
)
)
urlpatterns.append(
url(
r"^tests/data/content/code/filename/(?P<filename>.+)/$",
get_content_code_data_by_filename,
name="tests-content-code-filename",
)
)
urlpatterns.append(
url(
r"^tests/data/content/code/filenames/$",
get_content_code_data_all_filenames,
name="tests-content-code-filenames",
)
)
diff --git a/swh/web/tests/admin/test_origin_save.py b/swh/web/tests/admin/test_origin_save.py
index 0d3e1b07..3c3b4480 100644
--- a/swh/web/tests/admin/test_origin_save.py
+++ b/swh/web/tests/admin/test_origin_save.py
@@ -1,230 +1,231 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from urllib.parse import unquote
-from django.contrib.auth import get_user_model
import pytest
+from django.contrib.auth import get_user_model
+
from swh.web.common.models import (
SAVE_REQUEST_ACCEPTED,
SAVE_REQUEST_PENDING,
SAVE_REQUEST_REJECTED,
SAVE_TASK_NOT_YET_SCHEDULED,
SaveAuthorizedOrigin,
SaveOriginRequest,
SaveUnauthorizedOrigin,
)
from swh.web.common.origin_save import can_save_origin
from swh.web.common.utils import reverse
_user_name = "swh-web-admin"
_user_mail = "admin@swh-web.org"
_user_password = "..34~pounds~BEAUTY~march~63.."
_authorized_origin_url = "https://scm.ourproject.org/anonscm/"
_unauthorized_origin_url = "https://www.softwareheritage.org/"
pytestmark = pytest.mark.django_db
@pytest.fixture(autouse=True)
def populated_db():
User = get_user_model()
user = User.objects.create_user(_user_name, _user_mail, _user_password)
user.is_staff = True
user.save()
SaveAuthorizedOrigin.objects.create(url=_authorized_origin_url)
SaveUnauthorizedOrigin.objects.create(url=_unauthorized_origin_url)
def check_not_login(client, url):
login_url = reverse("login", query_params={"next": url})
response = client.post(url)
assert response.status_code == 302
assert unquote(response.url) == login_url
def test_add_authorized_origin_url(client):
authorized_url = "https://scm.adullact.net/anonscm/"
assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING
url = reverse(
"admin-origin-save-add-authorized-url", url_args={"origin_url": authorized_url}
)
check_not_login(client, url)
assert can_save_origin(authorized_url) == SAVE_REQUEST_PENDING
client.login(username=_user_name, password=_user_password)
response = client.post(url)
assert response.status_code == 200
assert can_save_origin(authorized_url) == SAVE_REQUEST_ACCEPTED
def test_remove_authorized_origin_url(client):
assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED
url = reverse(
"admin-origin-save-remove-authorized-url",
url_args={"origin_url": _authorized_origin_url},
)
check_not_login(client, url)
assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_ACCEPTED
client.login(username=_user_name, password=_user_password)
response = client.post(url)
assert response.status_code == 200
assert can_save_origin(_authorized_origin_url) == SAVE_REQUEST_PENDING
def test_add_unauthorized_origin_url(client):
unauthorized_url = "https://www.yahoo./"
assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING
url = reverse(
"admin-origin-save-add-unauthorized-url",
url_args={"origin_url": unauthorized_url},
)
check_not_login(client, url)
assert can_save_origin(unauthorized_url) == SAVE_REQUEST_PENDING
client.login(username=_user_name, password=_user_password)
response = client.post(url)
assert response.status_code == 200
assert can_save_origin(unauthorized_url) == SAVE_REQUEST_REJECTED
def test_remove_unauthorized_origin_url(client):
assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED
url = reverse(
"admin-origin-save-remove-unauthorized-url",
url_args={"origin_url": _unauthorized_origin_url},
)
check_not_login(client, url)
assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_REJECTED
client.login(username=_user_name, password=_user_password)
response = client.post(url)
assert response.status_code == 200
assert can_save_origin(_unauthorized_origin_url) == SAVE_REQUEST_PENDING
def test_accept_pending_save_request(client, mocker):
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler")
visit_type = "git"
origin_url = "https://v2.pikacode.com/bthate/botlib.git"
save_request_url = reverse(
"api-1-save-origin",
url_args={"visit_type": visit_type, "origin_url": origin_url},
)
response = client.post(
save_request_url, data={}, content_type="application/x-www-form-urlencoded"
)
assert response.status_code == 200
assert response.data["save_request_status"] == SAVE_REQUEST_PENDING
accept_request_url = reverse(
"admin-origin-save-request-accept",
url_args={"visit_type": visit_type, "origin_url": origin_url},
)
check_not_login(client, accept_request_url)
tasks_data = [
{
"priority": "high",
"policy": "oneshot",
"type": "load-git",
"arguments": {"kwargs": {"repo_url": origin_url}, "args": []},
"status": "next_run_not_scheduled",
"id": 1,
}
]
mock_scheduler.create_tasks.return_value = tasks_data
mock_scheduler.get_tasks.return_value = tasks_data
client.login(username=_user_name, password=_user_password)
response = client.post(accept_request_url)
assert response.status_code == 200
response = client.get(save_request_url)
assert response.status_code == 200
assert response.data[0]["save_request_status"] == SAVE_REQUEST_ACCEPTED
assert response.data[0]["save_task_status"] == SAVE_TASK_NOT_YET_SCHEDULED
def test_reject_pending_save_request(client, mocker):
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler")
visit_type = "git"
origin_url = "https://wikipedia.com"
save_request_url = reverse(
"api-1-save-origin",
url_args={"visit_type": visit_type, "origin_url": origin_url},
)
response = client.post(
save_request_url, data={}, content_type="application/x-www-form-urlencoded"
)
assert response.status_code == 200
assert response.data["save_request_status"] == SAVE_REQUEST_PENDING
reject_request_url = reverse(
"admin-origin-save-request-reject",
url_args={"visit_type": visit_type, "origin_url": origin_url},
)
check_not_login(client, reject_request_url)
client.login(username=_user_name, password=_user_password)
response = client.post(reject_request_url)
assert response.status_code == 200
tasks_data = [
{
"priority": "high",
"policy": "oneshot",
"type": "load-git",
"arguments": {"kwargs": {"repo_url": origin_url}, "args": []},
"status": "next_run_not_scheduled",
"id": 1,
}
]
mock_scheduler.create_tasks.return_value = tasks_data
mock_scheduler.get_tasks.return_value = tasks_data
response = client.get(save_request_url)
assert response.status_code == 200
assert response.data[0]["save_request_status"] == SAVE_REQUEST_REJECTED
def test_remove_save_request(client):
sor = SaveOriginRequest.objects.create(
visit_type="git",
origin_url="https://wikipedia.com",
status=SAVE_REQUEST_PENDING,
)
assert SaveOriginRequest.objects.count() == 1
remove_request_url = reverse(
"admin-origin-save-request-remove", url_args={"sor_id": sor.id}
)
check_not_login(client, remove_request_url)
client.login(username=_user_name, password=_user_password)
response = client.post(remove_request_url)
assert response.status_code == 200
assert SaveOriginRequest.objects.count() == 0
diff --git a/swh/web/tests/api/test_apidoc.py b/swh/web/tests/api/test_apidoc.py
index a99fb175..afdba2c1 100644
--- a/swh/web/tests/api/test_apidoc.py
+++ b/swh/web/tests/api/test_apidoc.py
@@ -1,493 +1,494 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import textwrap
import pytest
+
from rest_framework.response import Response
from swh.storage.exc import StorageAPIError, StorageDBError
from swh.web.api.apidoc import _parse_httpdomain_doc, api_doc
from swh.web.api.apiurls import api_route
from swh.web.common.exc import BadInputExc, ForbiddenExc, NotFoundExc
from swh.web.common.utils import prettify_html, reverse
from swh.web.tests.django_asserts import assert_template_used
_httpdomain_doc = """
.. http:get:: /api/1/revision/(sha1_git)/
Get information about a revision in the archive.
Revisions are identified by **sha1** checksums, compatible with Git commit
identifiers.
See :func:`swh.model.identifiers.revision_identifier` in our data model
module for details about how they are computed.
:param string sha1_git: hexadecimal representation of the revision
**sha1_git** identifier
:reqheader Accept: the requested response content type,
either ``application/json`` (default) or ``application/yaml``
:resheader Content-Type: this depends on :http:header:`Accept` header
of request
:<json int n: sample input integer
:<json string s: sample input string
:<json array a: sample input array
:>json object author: information about the author of the revision
:>json object committer: information about the committer of the revision
:>json string committer_date: ISO representation of the commit date
(in UTC)
:>json string date: ISO representation of the revision date (in UTC)
:>json string directory: the unique identifier that revision points to
:>json string directory_url: link to
:http:get:`/api/1/directory/(sha1_git)/[(path)/]` to get information
about the directory associated to the revision
:>json string id: the revision unique identifier
:>json boolean merge: whether or not the revision corresponds to a merge
commit
:>json string message: the message associated to the revision
:>json array parents: the parents of the revision, i.e. the previous
revisions that head directly to it, each entry of that array contains
an unique parent revision identifier but also a link to
:http:get:`/api/1/revision/(sha1_git)/` to get more information
about it
:>json string type: the type of the revision
:statuscode 200: no error
:statuscode 400: an invalid **sha1_git** value has been provided
:statuscode 404: requested revision can not be found in the archive
**Request:**
.. parsed-literal::
:swh_web_api:`revision/aafb16d69fd30ff58afdd69036a26047f3aebdc6/`
"""
_exception_http_code = {
BadInputExc: 400,
ForbiddenExc: 403,
NotFoundExc: 404,
Exception: 500,
StorageAPIError: 503,
StorageDBError: 503,
}
def test_apidoc_nodoc_failure():
with pytest.raises(Exception):
@api_doc("/my/nodoc/url/")
def apidoc_nodoc_tester(request, arga=0, argb=0):
return Response(arga + argb)
@api_route(r"/some/(?P<myarg>[0-9]+)/(?P<myotherarg>[0-9]+)/", "api-1-some-doc-route")
@api_doc("/some/doc/route/")
def apidoc_route(request, myarg, myotherarg, akw=0):
"""
Sample doc
"""
return {"result": int(myarg) + int(myotherarg) + akw}
def test_apidoc_route_doc(client):
url = reverse("api-1-some-doc-route-doc")
rv = client.get(url, HTTP_ACCEPT="text/html")
assert rv.status_code == 200, rv.content
assert_template_used(rv, "api/apidoc.html")
def test_apidoc_route_fn(api_client):
url = reverse("api-1-some-doc-route", url_args={"myarg": 1, "myotherarg": 1})
rv = api_client.get(url)
assert rv.status_code == 200, rv.data
@api_route(r"/test/error/(?P<exc_name>.+)/", "api-1-test-error")
@api_doc("/test/error/")
def apidoc_test_error_route(request, exc_name):
"""
Sample doc
"""
for e in _exception_http_code.keys():
if e.__name__ == exc_name:
raise e("Error")
def test_apidoc_error(api_client):
for exc, code in _exception_http_code.items():
url = reverse("api-1-test-error", url_args={"exc_name": exc.__name__})
rv = api_client.get(url)
assert rv.status_code == code, rv.data
@api_route(
r"/some/full/(?P<myarg>[0-9]+)/(?P<myotherarg>[0-9]+)/",
"api-1-some-complete-doc-route",
)
@api_doc("/some/complete/doc/route/")
def apidoc_full_stack(request, myarg, myotherarg, akw=0):
"""
Sample doc
"""
return {"result": int(myarg) + int(myotherarg) + akw}
def test_apidoc_full_stack_doc(client):
url = reverse("api-1-some-complete-doc-route-doc")
rv = client.get(url, HTTP_ACCEPT="text/html")
assert rv.status_code == 200, rv.content
assert_template_used(rv, "api/apidoc.html")
def test_apidoc_full_stack_fn(api_client):
url = reverse(
"api-1-some-complete-doc-route", url_args={"myarg": 1, "myotherarg": 1}
)
rv = api_client.get(url)
assert rv.status_code == 200, rv.data
@api_route(r"/test/post/only/", "api-1-test-post-only", methods=["POST"])
@api_doc("/test/post/only/")
def apidoc_test_post_only(request, exc_name):
"""
Sample doc
"""
return {"result": "some data"}
def test_apidoc_post_only(client):
# a dedicated view accepting GET requests should have
# been created to display the HTML documentation
url = reverse("api-1-test-post-only-doc")
rv = client.get(url, HTTP_ACCEPT="text/html")
assert rv.status_code == 200, rv.content
assert_template_used(rv, "api/apidoc.html")
def test_api_doc_parse_httpdomain():
doc_data = {
"description": "",
"urls": [],
"args": [],
"params": [],
"resheaders": [],
"reqheaders": [],
"input_type": "",
"inputs": [],
"return_type": "",
"returns": [],
"status_codes": [],
"examples": [],
}
_parse_httpdomain_doc(_httpdomain_doc, doc_data)
expected_urls = [
{
"rule": "/api/1/revision/ **\\(sha1_git\\)** /",
"methods": ["GET", "HEAD", "OPTIONS"],
}
]
assert "urls" in doc_data
assert doc_data["urls"] == expected_urls
expected_description = (
"Get information about a revision in the archive. "
"Revisions are identified by **sha1** checksums, "
"compatible with Git commit identifiers. See "
"**swh.model.identifiers.revision_identifier** in "
"our data model module for details about how they "
"are computed."
)
assert "description" in doc_data
assert doc_data["description"] == expected_description
expected_args = [
{
"name": "sha1_git",
"type": "string",
"doc": (
"hexadecimal representation of the revision " "**sha1_git** identifier"
),
}
]
assert "args" in doc_data
assert doc_data["args"] == expected_args
expected_params = []
assert "params" in doc_data
assert doc_data["params"] == expected_params
expected_reqheaders = [
{
"doc": (
"the requested response content type, either "
"``application/json`` (default) or ``application/yaml``"
),
"name": "Accept",
}
]
assert "reqheaders" in doc_data
assert doc_data["reqheaders"] == expected_reqheaders
expected_resheaders = [
{"doc": "this depends on **Accept** header of request", "name": "Content-Type"}
]
assert "resheaders" in doc_data
assert doc_data["resheaders"] == expected_resheaders
expected_statuscodes = [
{"code": "200", "doc": "no error"},
{"code": "400", "doc": "an invalid **sha1_git** value has been provided"},
{"code": "404", "doc": "requested revision can not be found in the archive"},
]
assert "status_codes" in doc_data
assert doc_data["status_codes"] == expected_statuscodes
expected_input_type = "object"
assert "input_type" in doc_data
assert doc_data["input_type"] == expected_input_type
expected_inputs = [
{"name": "n", "type": "int", "doc": "sample input integer"},
{"name": "s", "type": "string", "doc": "sample input string"},
{"name": "a", "type": "array", "doc": "sample input array"},
]
assert "inputs" in doc_data
assert doc_data["inputs"] == expected_inputs
expected_return_type = "object"
assert "return_type" in doc_data
assert doc_data["return_type"] == expected_return_type
expected_returns = [
{
"name": "author",
"type": "object",
"doc": "information about the author of the revision",
},
{
"name": "committer",
"type": "object",
"doc": "information about the committer of the revision",
},
{
"name": "committer_date",
"type": "string",
"doc": "ISO representation of the commit date (in UTC)",
},
{
"name": "date",
"type": "string",
"doc": "ISO representation of the revision date (in UTC)",
},
{
"name": "directory",
"type": "string",
"doc": "the unique identifier that revision points to",
},
{
"name": "directory_url",
"type": "string",
"doc": (
"link to `/api/1/directory/ </api/1/directory/doc/>`_ "
"to get information about the directory associated to "
"the revision"
),
},
{"name": "id", "type": "string", "doc": "the revision unique identifier"},
{
"name": "merge",
"type": "boolean",
"doc": "whether or not the revision corresponds to a merge commit",
},
{
"name": "message",
"type": "string",
"doc": "the message associated to the revision",
},
{
"name": "parents",
"type": "array",
"doc": (
"the parents of the revision, i.e. the previous revisions "
"that head directly to it, each entry of that array "
"contains an unique parent revision identifier but also a "
"link to `/api/1/revision/ </api/1/revision/doc/>`_ "
"to get more information about it"
),
},
{"name": "type", "type": "string", "doc": "the type of the revision"},
]
assert "returns" in doc_data
assert doc_data["returns"] == expected_returns
expected_examples = ["/api/1/revision/aafb16d69fd30ff58afdd69036a26047f3aebdc6/"]
assert "examples" in doc_data
assert doc_data["examples"] == expected_examples
@api_route(r"/post/endpoint/", "api-1-post-endpoint", methods=["POST"])
@api_doc("/post/endpoint/")
def apidoc_test_post_endpoint(request):
"""
.. http:post:: /api/1/post/endpoint/
Endpoint documentation
:<jsonarr string -: Input array of SWHIDs
:>json object <swhid>: an object whose keys are input SWHIDs
and values objects with the following keys:
* **known (bool)**: whether the object was found
"""
pass
def test_apidoc_input_output_doc(client):
url = reverse("api-1-post-endpoint-doc")
rv = client.get(url, HTTP_ACCEPT="text/html")
assert rv.status_code == 200, rv.content
assert_template_used(rv, "api/apidoc.html")
input_html_doc = textwrap.indent(
(
'<dl class="row">\n'
' <dt class="col col-md-2 text-right">\n'
" array\n"
" </dt>\n"
' <dd class="col col-md-9">\n'
" <p>\n"
" Input array of SWHIDs\n"
" </p>\n"
" </dd>\n"
"</dl>\n"
),
" " * 7,
)
output_html_doc = textwrap.indent(
(
'<dl class="row">\n'
' <dt class="col col-md-2 text-right">\n'
" object\n"
" </dt>\n"
' <dd class="col col-md-9">\n'
" <p>\n"
" an object containing the following keys:\n"
" </p>\n"
' <div class="swh-rst">\n'
" <blockquote>\n"
" <ul>\n"
" <li>\n"
" <p>\n"
" <strong>\n"
" <swhid> (object)\n"
" </strong>\n"
" : an object whose keys are input SWHIDs"
" and values objects with the following keys:\n"
" </p>\n"
" <blockquote>\n"
' <ul class="simple">\n'
" <li>\n"
" <p>\n"
" <strong>\n"
" known (bool)\n"
" </strong>\n"
" : whether the object was found\n"
" </p>\n"
" </li>\n"
" </ul>\n"
" </blockquote>\n"
" </li>\n"
" </ul>\n"
" </blockquote>\n"
" </div>\n"
" </dd>\n"
"</dl>\n"
),
" " * 7,
)
html = prettify_html(rv.content)
assert input_html_doc in html
assert output_html_doc in html
@api_route(r"/endpoint/links/in/doc/", "api-1-endpoint-links-in-doc")
@api_doc("/endpoint/links/in/doc/")
def apidoc_test_endpoint_with_links_in_doc(request):
"""
.. http:get:: /api/1/post/endpoint/
Endpoint documentation with links to
:http:get:`/api/1/content/[(hash_type):](hash)/`,
:http:get:`/api/1/directory/(sha1_git)/[(path)/]`
and `archive <https://archive.softwareheritage.org>`_.
"""
pass
def test_apidoc_with_links(client):
url = reverse("api-1-endpoint-links-in-doc")
rv = client.get(url, HTTP_ACCEPT="text/html")
assert rv.status_code == 200, rv.content
assert_template_used(rv, "api/apidoc.html")
html = prettify_html(rv.content)
first_link = textwrap.indent(
(
'<a class="reference external" href="/api/1/content/doc/">\n'
" /api/1/content/\n"
"</a>"
),
" " * 9,
)
second_link = textwrap.indent(
(
'<a class="reference external" href="/api/1/directory/doc/">\n'
" /api/1/directory/\n"
"</a>"
),
" " * 9,
)
third_link = textwrap.indent(
(
'<a class="reference external" '
'href="https:/archive.softwareheritage.org">\n'
" archive\n"
"</a>"
),
" " * 9,
)
assert first_link in html
assert second_link in html
assert third_link in html
diff --git a/swh/web/tests/api/test_throttling.py b/swh/web/tests/api/test_throttling.py
index d3244440..79818390 100644
--- a/swh/web/tests/api/test_throttling.py
+++ b/swh/web/tests/api/test_throttling.py
@@ -1,223 +1,224 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import pytest
+
from django.conf.urls import url
from django.contrib.auth.models import Permission, User
from django.contrib.contenttypes.models import ContentType
from django.test.utils import override_settings
-import pytest
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.views import APIView
from swh.web.api.throttling import (
API_THROTTLING_EXEMPTED_PERM,
SwhWebRateThrottle,
throttle_scope,
)
from swh.web.settings.tests import (
scope1_limiter_rate,
scope1_limiter_rate_post,
scope2_limiter_rate,
scope2_limiter_rate_post,
scope3_limiter_rate,
scope3_limiter_rate_post,
)
from swh.web.urls import urlpatterns
class MockViewScope1(APIView):
throttle_classes = (SwhWebRateThrottle,)
throttle_scope = "scope1"
def get(self, request):
return Response("foo_get")
def post(self, request):
return Response("foo_post")
@api_view(["GET", "POST"])
@throttle_scope("scope2")
def mock_view_scope2(request):
if request.method == "GET":
return Response("bar_get")
elif request.method == "POST":
return Response("bar_post")
class MockViewScope3(APIView):
throttle_classes = (SwhWebRateThrottle,)
throttle_scope = "scope3"
def get(self, request):
return Response("foo_get")
def post(self, request):
return Response("foo_post")
@api_view(["GET", "POST"])
@throttle_scope("scope3")
def mock_view_scope3(request):
if request.method == "GET":
return Response("bar_get")
elif request.method == "POST":
return Response("bar_post")
urlpatterns += [
url(r"^scope1_class$", MockViewScope1.as_view()),
url(r"^scope2_func$", mock_view_scope2),
url(r"^scope3_class$", MockViewScope3.as_view()),
url(r"^scope3_func$", mock_view_scope3),
]
def check_response(response, status_code, limit=None, remaining=None):
assert response.status_code == status_code
if limit is not None:
assert response["X-RateLimit-Limit"] == str(limit)
else:
assert "X-RateLimit-Limit" not in response
if remaining is not None:
assert response["X-RateLimit-Remaining"] == str(remaining)
else:
assert "X-RateLimit-Remaining" not in response
@override_settings(ROOT_URLCONF=__name__)
def test_scope1_requests_are_throttled(api_client):
"""
Ensure request rate is limited in scope1
"""
for i in range(scope1_limiter_rate):
response = api_client.get("/scope1_class")
check_response(response, 200, scope1_limiter_rate, scope1_limiter_rate - i - 1)
response = api_client.get("/scope1_class")
check_response(response, 429, scope1_limiter_rate, 0)
for i in range(scope1_limiter_rate_post):
response = api_client.post("/scope1_class")
check_response(
response, 200, scope1_limiter_rate_post, scope1_limiter_rate_post - i - 1
)
response = api_client.post("/scope1_class")
check_response(response, 429, scope1_limiter_rate_post, 0)
@override_settings(ROOT_URLCONF=__name__)
def test_scope2_requests_are_throttled(api_client):
"""
Ensure request rate is limited in scope2
"""
for i in range(scope2_limiter_rate):
response = api_client.get("/scope2_func")
check_response(response, 200, scope2_limiter_rate, scope2_limiter_rate - i - 1)
response = api_client.get("/scope2_func")
check_response(response, 429, scope2_limiter_rate, 0)
for i in range(scope2_limiter_rate_post):
response = api_client.post("/scope2_func")
check_response(
response, 200, scope2_limiter_rate_post, scope2_limiter_rate_post - i - 1
)
response = api_client.post("/scope2_func")
check_response(response, 429, scope2_limiter_rate_post, 0)
@override_settings(ROOT_URLCONF=__name__)
def test_scope3_requests_are_throttled_exempted(api_client):
"""
Ensure request rate is not limited in scope3 as
requests coming from localhost are exempted from rate limit.
"""
for _ in range(scope3_limiter_rate + 1):
response = api_client.get("/scope3_class")
check_response(response, 200)
for _ in range(scope3_limiter_rate_post + 1):
response = api_client.post("/scope3_class")
check_response(response, 200)
for _ in range(scope3_limiter_rate + 1):
response = api_client.get("/scope3_func")
check_response(response, 200)
for _ in range(scope3_limiter_rate_post + 1):
response = api_client.post("/scope3_func")
check_response(response, 200)
@override_settings(ROOT_URLCONF=__name__)
@pytest.mark.django_db
def test_staff_users_are_not_rate_limited(api_client):
staff_user = User.objects.create_user(
username="johndoe", password="", is_staff=True
)
api_client.force_login(staff_user)
for _ in range(scope2_limiter_rate + 1):
response = api_client.get("/scope2_func")
check_response(response, 200)
for _ in range(scope2_limiter_rate_post + 1):
response = api_client.post("/scope2_func")
check_response(response, 200)
@override_settings(ROOT_URLCONF=__name__)
@pytest.mark.django_db
def test_non_staff_users_are_rate_limited(api_client):
user = User.objects.create_user(username="johndoe", password="", is_staff=False)
api_client.force_login(user)
for i in range(scope2_limiter_rate):
response = api_client.get("/scope2_func")
check_response(response, 200, scope2_limiter_rate, scope2_limiter_rate - i - 1)
response = api_client.get("/scope2_func")
check_response(response, 429, scope2_limiter_rate, 0)
for i in range(scope2_limiter_rate_post):
response = api_client.post("/scope2_func")
check_response(
response, 200, scope2_limiter_rate_post, scope2_limiter_rate_post - i - 1
)
response = api_client.post("/scope2_func")
check_response(response, 429, scope2_limiter_rate_post, 0)
@override_settings(ROOT_URLCONF=__name__)
@pytest.mark.django_db
def test_users_with_throttling_exempted_perm_are_not_rate_limited(api_client):
user = User.objects.create_user(username="johndoe", password="")
perm_splitted = API_THROTTLING_EXEMPTED_PERM.split(".")
app_label = ".".join(perm_splitted[:-1])
perm_name = perm_splitted[-1]
content_type = ContentType.objects.create(app_label=app_label, model="dummy")
permission = Permission.objects.create(
codename=perm_name, name=perm_name, content_type=content_type,
)
user.user_permissions.add(permission)
assert user.has_perm(API_THROTTLING_EXEMPTED_PERM)
api_client.force_login(user)
for _ in range(scope2_limiter_rate + 1):
response = api_client.get("/scope2_func")
check_response(response, 200)
for _ in range(scope2_limiter_rate_post + 1):
response = api_client.post("/scope2_func")
check_response(response, 200)
diff --git a/swh/web/tests/api/views/test_origin_save.py b/swh/web/tests/api/views/test_origin_save.py
index 2e063856..c3c20ddb 100644
--- a/swh/web/tests/api/views/test_origin_save.py
+++ b/swh/web/tests/api/views/test_origin_save.py
@@ -1,320 +1,321 @@
# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime, timedelta
-from django.utils import timezone
import pytest
+from django.utils import timezone
+
from swh.web.common.models import (
SAVE_REQUEST_ACCEPTED,
SAVE_REQUEST_PENDING,
SAVE_REQUEST_REJECTED,
SAVE_TASK_FAILED,
SAVE_TASK_NOT_CREATED,
SAVE_TASK_NOT_YET_SCHEDULED,
SAVE_TASK_SCHEDULED,
SAVE_TASK_SUCCEED,
SaveOriginRequest,
SaveUnauthorizedOrigin,
)
from swh.web.common.utils import reverse
from swh.web.tests.api.views import check_api_get_responses, check_api_post_responses
pytestmark = pytest.mark.django_db
@pytest.fixture(autouse=True)
def populated_db():
SaveUnauthorizedOrigin.objects.create(url="https://github.com/user/illegal_repo")
SaveUnauthorizedOrigin.objects.create(url="https://gitlab.com/user_to_exclude")
def test_invalid_visit_type(api_client):
url = reverse(
"api-1-save-origin",
url_args={
"visit_type": "foo",
"origin_url": "https://github.com/torvalds/linux",
},
)
check_api_get_responses(api_client, url, status_code=400)
def test_invalid_origin_url(api_client):
url = reverse(
"api-1-save-origin", url_args={"visit_type": "git", "origin_url": "bar"}
)
check_api_get_responses(api_client, url, status_code=400)
def check_created_save_request_status(
api_client,
mocker,
origin_url,
scheduler_task_status,
expected_request_status,
expected_task_status=None,
visit_date=None,
):
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler")
if not scheduler_task_status:
mock_scheduler.get_tasks.return_value = []
else:
mock_scheduler.get_tasks.return_value = [
{
"priority": "high",
"policy": "oneshot",
"type": "load-git",
"arguments": {"kwargs": {"repo_url": origin_url}, "args": []},
"status": scheduler_task_status,
"id": 1,
}
]
mock_scheduler.create_tasks.return_value = [
{
"priority": "high",
"policy": "oneshot",
"type": "load-git",
"arguments": {"kwargs": {"repo_url": origin_url}, "args": []},
"status": "next_run_not_scheduled",
"id": 1,
}
]
url = reverse(
"api-1-save-origin", url_args={"visit_type": "git", "origin_url": origin_url}
)
mock_visit_date = mocker.patch(
("swh.web.common.origin_save." "_get_visit_info_for_save_request")
)
mock_visit_date.return_value = (visit_date, None)
if expected_request_status != SAVE_REQUEST_REJECTED:
response = check_api_post_responses(api_client, url, data=None, status_code=200)
assert response.data["save_request_status"] == expected_request_status
assert response.data["save_task_status"] == expected_task_status
else:
check_api_post_responses(api_client, url, data=None, status_code=403)
def check_save_request_status(
api_client,
mocker,
origin_url,
expected_request_status,
expected_task_status,
scheduler_task_status="next_run_not_scheduled",
visit_date=None,
):
mock_scheduler = mocker.patch("swh.web.common.origin_save.scheduler")
mock_scheduler.get_tasks.return_value = [
{
"priority": "high",
"policy": "oneshot",
"type": "load-git",
"arguments": {"kwargs": {"repo_url": origin_url}, "args": []},
"status": scheduler_task_status,
"id": 1,
}
]
url = reverse(
"api-1-save-origin", url_args={"visit_type": "git", "origin_url": origin_url}
)
mock_visit_date = mocker.patch(
("swh.web.common.origin_save." "_get_visit_info_for_save_request")
)
mock_visit_date.return_value = (visit_date, None)
response = check_api_get_responses(api_client, url, status_code=200)
save_request_data = response.data[0]
assert save_request_data["save_request_status"] == expected_request_status
assert save_request_data["save_task_status"] == expected_task_status
# Check that save task status is still available when
# the scheduler task has been archived
mock_scheduler.get_tasks.return_value = []
response = check_api_get_responses(api_client, url, status_code=200)
save_request_data = response.data[0]
assert save_request_data["save_task_status"] == expected_task_status
def test_save_request_rejected(api_client, mocker):
origin_url = "https://github.com/user/illegal_repo"
check_created_save_request_status(
api_client, mocker, origin_url, None, SAVE_REQUEST_REJECTED
)
check_save_request_status(
api_client, mocker, origin_url, SAVE_REQUEST_REJECTED, SAVE_TASK_NOT_CREATED
)
def test_save_request_pending(api_client, mocker):
origin_url = "https://unkwownforge.com/user/repo"
check_created_save_request_status(
api_client,
mocker,
origin_url,
None,
SAVE_REQUEST_PENDING,
SAVE_TASK_NOT_CREATED,
)
check_save_request_status(
api_client, mocker, origin_url, SAVE_REQUEST_PENDING, SAVE_TASK_NOT_CREATED
)
def test_save_request_succeed(api_client, mocker):
origin_url = "https://github.com/Kitware/CMake"
check_created_save_request_status(
api_client,
mocker,
origin_url,
None,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
)
check_save_request_status(
api_client,
mocker,
origin_url,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_SCHEDULED,
scheduler_task_status="next_run_scheduled",
)
check_save_request_status(
api_client,
mocker,
origin_url,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_SUCCEED,
scheduler_task_status="completed",
visit_date=None,
)
visit_date = datetime.now(tz=timezone.utc) + timedelta(hours=1)
check_save_request_status(
api_client,
mocker,
origin_url,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_SUCCEED,
scheduler_task_status="completed",
visit_date=visit_date,
)
def test_save_request_failed(api_client, mocker):
origin_url = "https://gitlab.com/inkscape/inkscape"
check_created_save_request_status(
api_client,
mocker,
origin_url,
None,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
)
check_save_request_status(
api_client,
mocker,
origin_url,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_SCHEDULED,
scheduler_task_status="next_run_scheduled",
)
check_save_request_status(
api_client,
mocker,
origin_url,
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_FAILED,
scheduler_task_status="disabled",
)
def test_create_save_request_only_when_needed(api_client, mocker):
origin_url = "https://github.com/webpack/webpack"
SaveOriginRequest.objects.create(
visit_type="git",
origin_url=origin_url,
status=SAVE_REQUEST_ACCEPTED,
loading_task_id=56,
)
check_created_save_request_status(
api_client,
mocker,
origin_url,
"next_run_not_scheduled",
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
)
sors = list(
SaveOriginRequest.objects.filter(visit_type="git", origin_url=origin_url)
)
assert len(sors) == 1
check_created_save_request_status(
api_client,
mocker,
origin_url,
"next_run_scheduled",
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_SCHEDULED,
)
sors = list(
SaveOriginRequest.objects.filter(visit_type="git", origin_url=origin_url)
)
assert len(sors) == 1
visit_date = datetime.now(tz=timezone.utc) + timedelta(hours=1)
check_created_save_request_status(
api_client,
mocker,
origin_url,
"completed",
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
visit_date=visit_date,
)
sors = list(
SaveOriginRequest.objects.filter(visit_type="git", origin_url=origin_url)
)
# check_api_post_responses sends two POST requests to check YAML and JSON response
assert len(sors) == 3
check_created_save_request_status(
api_client,
mocker,
origin_url,
"disabled",
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
)
sors = list(
SaveOriginRequest.objects.filter(visit_type="git", origin_url=origin_url)
)
assert len(sors) == 5
def test_get_save_requests_unknown_origin(api_client):
unknown_origin_url = "https://gitlab.com/foo/bar"
url = reverse(
"api-1-save-origin",
url_args={"visit_type": "git", "origin_url": unknown_origin_url},
)
response = check_api_get_responses(api_client, url, status_code=404)
assert response.data == {
"exception": "NotFoundExc",
"reason": (
"No save requests found for visit of type " "git on origin with url %s."
)
% unknown_origin_url,
}
diff --git a/swh/web/tests/auth/test_api_auth.py b/swh/web/tests/auth/test_api_auth.py
index 654b8cb6..f05ec593 100644
--- a/swh/web/tests/auth/test_api_auth.py
+++ b/swh/web/tests/auth/test_api_auth.py
@@ -1,113 +1,114 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
-from django.contrib.auth.models import AnonymousUser, User
import pytest
+from django.contrib.auth.models import AnonymousUser, User
+
from swh.web.auth.models import OIDCUser
from swh.web.common.utils import reverse
from . import sample_data
from .keycloak_mock import mock_keycloak
@pytest.mark.django_db
def test_drf_django_session_auth_success(mocker, client):
"""
Check user gets authenticated when querying the web api
through a web browser.
"""
url = reverse("api-1-stat-counters")
mock_keycloak(mocker)
client.login(code="", code_verifier="", redirect_uri="")
response = client.get(url)
request = response.wsgi_request
assert response.status_code == 200
# user should be authenticated
assert isinstance(request.user, OIDCUser)
# check remoter used has not been saved to Django database
with pytest.raises(User.DoesNotExist):
User.objects.get(username=request.user.username)
@pytest.mark.django_db
def test_drf_oidc_bearer_token_auth_success(mocker, api_client):
"""
Check user gets authenticated when querying the web api
through an HTTP client using bearer token authentication.
"""
url = reverse("api-1-stat-counters")
refresh_token = sample_data.oidc_profile["refresh_token"]
mock_keycloak(mocker)
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {refresh_token}")
response = api_client.get(url)
request = response.wsgi_request
assert response.status_code == 200
# user should be authenticated
assert isinstance(request.user, OIDCUser)
# check remoter used has not been saved to Django database
with pytest.raises(User.DoesNotExist):
User.objects.get(username=request.user.username)
@pytest.mark.django_db
def test_drf_oidc_bearer_token_auth_failure(mocker, api_client):
url = reverse("api-1-stat-counters")
refresh_token = sample_data.oidc_profile["refresh_token"]
# check for failed authentication but with expected token format
mock_keycloak(mocker, auth_success=False)
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {refresh_token}")
response = api_client.get(url)
request = response.wsgi_request
assert response.status_code == 403
assert isinstance(request.user, AnonymousUser)
# check for failed authentication when token format is invalid
api_client.credentials(HTTP_AUTHORIZATION="Bearer invalid-token-format-ééà à ")
response = api_client.get(url)
request = response.wsgi_request
assert response.status_code == 400
assert isinstance(request.user, AnonymousUser)
def test_drf_oidc_auth_invalid_or_missing_authorization_type(api_client):
url = reverse("api-1-stat-counters")
refresh_token = sample_data.oidc_profile["refresh_token"]
# missing authorization type
api_client.credentials(HTTP_AUTHORIZATION=f"{refresh_token}")
response = api_client.get(url)
request = response.wsgi_request
assert response.status_code == 403
assert isinstance(request.user, AnonymousUser)
# invalid authorization type
api_client.credentials(HTTP_AUTHORIZATION="Foo token")
response = api_client.get(url)
request = response.wsgi_request
assert response.status_code == 403
assert isinstance(request.user, AnonymousUser)
diff --git a/swh/web/tests/auth/test_backends.py b/swh/web/tests/auth/test_backends.py
index ca03f160..24fcdf92 100644
--- a/swh/web/tests/auth/test_backends.py
+++ b/swh/web/tests/auth/test_backends.py
@@ -1,205 +1,206 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime, timedelta
+import pytest
+
from django.conf import settings
from django.contrib.auth import authenticate, get_backends
-import pytest
from rest_framework.exceptions import AuthenticationFailed
from swh.web.auth.backends import OIDCBearerTokenAuthentication
from swh.web.auth.models import OIDCUser
from swh.web.common.utils import reverse
from . import sample_data
from .keycloak_mock import mock_keycloak
def _authenticate_user(request_factory):
request = request_factory.get(reverse("oidc-login-complete"))
return authenticate(
request=request,
code="some-code",
code_verifier="some-code-verifier",
redirect_uri="https://localhost:5004",
)
def _check_authenticated_user(user, decoded_token, kc_oidc_mock):
assert user is not None
assert isinstance(user, OIDCUser)
assert user.id != 0
assert user.username == decoded_token["preferred_username"]
assert user.password == ""
assert user.first_name == decoded_token["given_name"]
assert user.last_name == decoded_token["family_name"]
assert user.email == decoded_token["email"]
assert user.is_staff == ("/staff" in decoded_token["groups"])
assert user.sub == decoded_token["sub"]
resource_access = decoded_token.get("resource_access", {})
resource_access_client = resource_access.get(kc_oidc_mock, {})
assert user.permissions == set(resource_access_client.get("roles", []))
@pytest.mark.django_db
def test_oidc_code_pkce_auth_backend_success(mocker, request_factory):
"""
Checks successful login based on OpenID Connect with PKCE extension
Django authentication backend (login from Web UI).
"""
kc_oidc_mock = mock_keycloak(mocker, user_groups=["/staff"])
oidc_profile = sample_data.oidc_profile
user = _authenticate_user(request_factory)
decoded_token = kc_oidc_mock.decode_token(user.access_token)
_check_authenticated_user(user, decoded_token, kc_oidc_mock)
auth_datetime = datetime.fromtimestamp(decoded_token["auth_time"])
exp_datetime = datetime.fromtimestamp(decoded_token["exp"])
refresh_exp_datetime = auth_datetime + timedelta(
seconds=oidc_profile["refresh_expires_in"]
)
assert user.access_token == oidc_profile["access_token"]
assert user.expires_at == exp_datetime
assert user.id_token == oidc_profile["id_token"]
assert user.refresh_token == oidc_profile["refresh_token"]
assert user.refresh_expires_at == refresh_exp_datetime
assert user.scope == oidc_profile["scope"]
assert user.session_state == oidc_profile["session_state"]
backend_path = "swh.web.auth.backends.OIDCAuthorizationCodePKCEBackend"
assert user.backend == backend_path
backend_idx = settings.AUTHENTICATION_BACKENDS.index(backend_path)
assert get_backends()[backend_idx].get_user(user.id) == user
@pytest.mark.django_db
def test_oidc_code_pkce_auth_backend_failure(mocker, request_factory):
"""
Checks failed login based on OpenID Connect with PKCE extension Django
authentication backend (login from Web UI).
"""
mock_keycloak(mocker, auth_success=False)
user = _authenticate_user(request_factory)
assert user is None
@pytest.mark.django_db
def test_oidc_code_pkce_auth_backend_permissions(mocker, request_factory):
"""
Checks that a permission defined with OpenID Connect is correctly mapped
to a Django one when logging from Web UI.
"""
permission = "webapp.some-permission"
mock_keycloak(mocker, user_permissions=[permission])
user = _authenticate_user(request_factory)
assert user.has_perm(permission)
assert user.get_all_permissions() == {permission}
assert user.get_group_permissions() == {permission}
assert user.has_module_perms("webapp")
assert not user.has_module_perms("foo")
@pytest.mark.django_db
def test_drf_oidc_bearer_token_auth_backend_success(mocker, api_request_factory):
"""
Checks successful login based on OpenID Connect bearer token Django REST
Framework authentication backend (Web API login).
"""
url = reverse("api-1-stat-counters")
drf_auth_backend = OIDCBearerTokenAuthentication()
kc_oidc_mock = mock_keycloak(mocker)
refresh_token = sample_data.oidc_profile["refresh_token"]
access_token = sample_data.oidc_profile["access_token"]
decoded_token = kc_oidc_mock.decode_token(access_token)
request = api_request_factory.get(url, HTTP_AUTHORIZATION=f"Bearer {refresh_token}")
user, _ = drf_auth_backend.authenticate(request)
_check_authenticated_user(user, decoded_token, kc_oidc_mock)
# oidc_profile is not filled when authenticating through bearer token
assert hasattr(user, "access_token") and user.access_token is None
@pytest.mark.django_db
def test_drf_oidc_bearer_token_auth_backend_failure(mocker, api_request_factory):
"""
Checks failed login based on OpenID Connect bearer token Django REST
Framework authentication backend (Web API login).
"""
url = reverse("api-1-stat-counters")
drf_auth_backend = OIDCBearerTokenAuthentication()
# simulate a failed authentication with a bearer token in expected format
mock_keycloak(mocker, auth_success=False)
refresh_token = sample_data.oidc_profile["refresh_token"]
request = api_request_factory.get(url, HTTP_AUTHORIZATION=f"Bearer {refresh_token}")
with pytest.raises(AuthenticationFailed):
drf_auth_backend.authenticate(request)
# simulate a failed authentication with an invalid bearer token format
request = api_request_factory.get(
url, HTTP_AUTHORIZATION="Bearer invalid-token-format"
)
with pytest.raises(AuthenticationFailed):
drf_auth_backend.authenticate(request)
def test_drf_oidc_auth_invalid_or_missing_auth_type(api_request_factory):
"""
Checks failed login based on OpenID Connect bearer token Django REST
Framework authentication backend (Web API login) due to invalid
authorization header value.
"""
url = reverse("api-1-stat-counters")
drf_auth_backend = OIDCBearerTokenAuthentication()
refresh_token = sample_data.oidc_profile["refresh_token"]
# Invalid authorization type
request = api_request_factory.get(url, HTTP_AUTHORIZATION="Foo token")
with pytest.raises(AuthenticationFailed):
drf_auth_backend.authenticate(request)
# Missing authorization type
request = api_request_factory.get(url, HTTP_AUTHORIZATION=f"{refresh_token}")
with pytest.raises(AuthenticationFailed):
drf_auth_backend.authenticate(request)
@pytest.mark.django_db
def test_drf_oidc_bearer_token_auth_backend_permissions(mocker, api_request_factory):
"""
Checks that a permission defined with OpenID Connect is correctly mapped
to a Django one when using bearer token authentication.
"""
permission = "webapp.some-permission"
mock_keycloak(mocker, user_permissions=[permission])
drf_auth_backend = OIDCBearerTokenAuthentication()
refresh_token = sample_data.oidc_profile["refresh_token"]
url = reverse("api-1-stat-counters")
request = api_request_factory.get(url, HTTP_AUTHORIZATION=f"Bearer {refresh_token}")
user, _ = drf_auth_backend.authenticate(request)
assert user.has_perm(permission)
assert user.get_all_permissions() == {permission}
assert user.get_group_permissions() == {permission}
assert user.has_module_perms("webapp")
assert not user.has_module_perms("foo")
diff --git a/swh/web/tests/auth/test_middlewares.py b/swh/web/tests/auth/test_middlewares.py
index cc587128..636cb7b6 100644
--- a/swh/web/tests/auth/test_middlewares.py
+++ b/swh/web/tests/auth/test_middlewares.py
@@ -1,47 +1,48 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime
-from django.test import modify_settings
import pytest
+from django.test import modify_settings
+
from swh.web.common.utils import reverse
from .keycloak_mock import mock_keycloak
@pytest.mark.django_db
@modify_settings(
MIDDLEWARE={"remove": ["swh.web.auth.middlewares.OIDCSessionRefreshMiddleware"]}
)
def test_oidc_session_refresh_middleware_disabled(client, mocker):
# authenticate but make session expires immediately
kc_oidc_mock = mock_keycloak(mocker, exp=int(datetime.now().timestamp()))
client.login(code="", code_verifier="", redirect_uri="")
kc_oidc_mock.authorization_code.assert_called()
url = reverse("swh-web-homepage")
resp = client.get(url)
# no redirection for silent refresh
assert resp.status_code != 302
@pytest.mark.django_db
def test_oidc_session_refresh_middleware_enabled(client, mocker):
# authenticate but make session expires immediately
kc_oidc_mock = mock_keycloak(mocker, exp=int(datetime.now().timestamp()))
client.login(code="", code_verifier="", redirect_uri="")
kc_oidc_mock.authorization_code.assert_called()
url = reverse("swh-web-homepage")
resp = client.get(url)
# should redirect for silent session refresh
assert resp.status_code == 302
silent_refresh_url = reverse(
"oidc-login", query_params={"next_path": url, "prompt": "none"}
)
assert resp["location"] == silent_refresh_url
diff --git a/swh/web/tests/auth/test_views.py b/swh/web/tests/auth/test_views.py
index f9e7cc86..d58fbf53 100644
--- a/swh/web/tests/auth/test_views.py
+++ b/swh/web/tests/auth/test_views.py
@@ -1,337 +1,338 @@
# Copyright (C) 2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from urllib.parse import urljoin, urlparse
import uuid
+import pytest
+
from django.contrib.auth.models import AnonymousUser, User
from django.http import QueryDict
-import pytest
from swh.web.auth.models import OIDCUser
from swh.web.auth.utils import OIDC_SWH_WEB_CLIENT_ID
from swh.web.common.utils import reverse
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.urls import _default_view as homepage_view
from . import sample_data
from .keycloak_mock import mock_keycloak
@pytest.mark.django_db
def test_oidc_login_views_success(client, mocker):
"""
Simulate a successful login authentication with OpenID Connect
authorization code flow with PKCE.
"""
# mock Keycloak client
kc_oidc_mock = mock_keycloak(mocker)
# user initiates login process
login_url = reverse("oidc-login")
response = client.get(login_url)
request = response.wsgi_request
# should redirect to Keycloak authentication page in order
# for a user to login with its username / password
assert response.status_code == 302
assert isinstance(request.user, AnonymousUser)
parsed_url = urlparse(response["location"])
authorization_url = kc_oidc_mock.well_known()["authorization_endpoint"]
query_dict = QueryDict(parsed_url.query)
# check redirect url is valid
assert urljoin(response["location"], parsed_url.path) == authorization_url
assert "client_id" in query_dict
assert query_dict["client_id"] == OIDC_SWH_WEB_CLIENT_ID
assert "response_type" in query_dict
assert query_dict["response_type"] == "code"
assert "redirect_uri" in query_dict
assert query_dict["redirect_uri"] == reverse("oidc-login-complete", request=request)
assert "code_challenge_method" in query_dict
assert query_dict["code_challenge_method"] == "S256"
assert "scope" in query_dict
assert query_dict["scope"] == "openid"
assert "state" in query_dict
assert "code_challenge" in query_dict
# check a login_data has been registered in user session
assert "login_data" in request.session
login_data = request.session["login_data"]
assert "code_verifier" in login_data
assert "state" in login_data
assert "redirect_uri" in login_data
assert login_data["redirect_uri"] == query_dict["redirect_uri"]
# once a user has identified himself in Keycloak, he is
# redirected to the 'oidc-login-complete' view to
# login in Django.
# generate authorization code / session state in the same
# manner as Keycloak
code = f"{str(uuid.uuid4())}.{str(uuid.uuid4())}.{str(uuid.uuid4())}"
session_state = str(uuid.uuid4())
login_complete_url = reverse(
"oidc-login-complete",
query_params={
"code": code,
"state": login_data["state"],
"session_state": session_state,
},
)
# login process finalization
response = client.get(login_complete_url)
request = response.wsgi_request
# should redirect to root url by default
assert response.status_code == 302
assert response["location"] == request.build_absolute_uri("/")
# user should be authenticated
assert isinstance(request.user, OIDCUser)
# check remote user has not been saved to Django database
with pytest.raises(User.DoesNotExist):
User.objects.get(username=request.user.username)
@pytest.mark.django_db
def test_oidc_logout_view_success(client, mocker):
"""
Simulate a successful logout operation with OpenID Connect.
"""
# mock Keycloak client
kc_oidc_mock = mock_keycloak(mocker)
# login our test user
client.login(code="", code_verifier="", redirect_uri="")
kc_oidc_mock.authorization_code.assert_called()
# user initiates logout
oidc_logout_url = reverse("oidc-logout")
response = client.get(oidc_logout_url)
request = response.wsgi_request
# should redirect to logout page
assert response.status_code == 302
logout_url = reverse("logout", query_params={"remote_user": 1})
assert response["location"] == request.build_absolute_uri(logout_url)
# should have been logged out in Keycloak
kc_oidc_mock.logout.assert_called_with(sample_data.oidc_profile["refresh_token"])
# check effective logout in Django
assert isinstance(request.user, AnonymousUser)
@pytest.mark.django_db
def test_oidc_login_view_failure(client, mocker):
"""
Simulate a failed authentication with OpenID Connect.
"""
# mock Keycloak client
mock_keycloak(mocker, auth_success=False)
# user initiates login process
login_url = reverse("oidc-login")
response = client.get(login_url)
request = response.wsgi_request
# should render an error page
assert response.status_code == 500
assert_template_used(response, "error.html")
# no users should be logged in
assert isinstance(request.user, AnonymousUser)
# Simulate possible errors with OpenID Connect in the login complete view.
def test_oidc_login_complete_view_no_login_data(client, mocker):
# user initiates login process
login_url = reverse("oidc-login-complete")
response = client.get(login_url)
# should render an error page
assert_template_used(response, "error.html")
assert_contains(
response, "Login process has not been initialized.", status_code=500
)
def test_oidc_login_complete_view_missing_parameters(client, mocker):
# simulate login process has been initialized
session = client.session
session["login_data"] = {
"code_verifier": "",
"state": str(uuid.uuid4()),
"redirect_uri": "",
"next_path": "",
"prompt": "",
}
session.save()
# user initiates login process
login_url = reverse("oidc-login-complete")
response = client.get(login_url)
request = response.wsgi_request
# should render an error page
assert_template_used(response, "error.html")
assert_contains(
response, "Missing query parameters for authentication.", status_code=400
)
# no user should be logged in
assert isinstance(request.user, AnonymousUser)
def test_oidc_login_complete_wrong_csrf_token(client, mocker):
# mock Keycloak client
mock_keycloak(mocker)
# simulate login process has been initialized
session = client.session
session["login_data"] = {
"code_verifier": "",
"state": str(uuid.uuid4()),
"redirect_uri": "",
"next_path": "",
"prompt": "",
}
session.save()
# user initiates login process
login_url = reverse(
"oidc-login-complete", query_params={"code": "some-code", "state": "some-state"}
)
response = client.get(login_url)
request = response.wsgi_request
# should render an error page
assert_template_used(response, "error.html")
assert_contains(
response, "Wrong CSRF token, aborting login process.", status_code=400
)
# no user should be logged in
assert isinstance(request.user, AnonymousUser)
@pytest.mark.django_db
def test_oidc_login_complete_wrong_code_verifier(client, mocker):
# mock Keycloak client
mock_keycloak(mocker, auth_success=False)
# simulate login process has been initialized
session = client.session
session["login_data"] = {
"code_verifier": "",
"state": str(uuid.uuid4()),
"redirect_uri": "",
"next_path": "",
"prompt": "",
}
session.save()
# check authentication error is reported
login_url = reverse(
"oidc-login-complete",
query_params={"code": "some-code", "state": session["login_data"]["state"]},
)
response = client.get(login_url)
request = response.wsgi_request
# should render an error page
assert_template_used(response, "error.html")
assert_contains(response, "User authentication failed.", status_code=500)
# no user should be logged in
assert isinstance(request.user, AnonymousUser)
@pytest.mark.django_db
def test_oidc_logout_view_failure(client, mocker):
"""
Simulate a failed logout operation with OpenID Connect.
"""
# mock Keycloak client
kc_oidc_mock = mock_keycloak(mocker)
# login our test user
client.login(code="", code_verifier="", redirect_uri="")
err_msg = "Authentication server error"
kc_oidc_mock.logout.side_effect = Exception(err_msg)
# user initiates logout process
logout_url = reverse("oidc-logout")
response = client.get(logout_url)
request = response.wsgi_request
# should render an error page
assert_template_used(response, "error.html")
assert_contains(response, err_msg, status_code=500)
# user should be logged out from Django anyway
assert isinstance(request.user, AnonymousUser)
@pytest.mark.django_db
def test_oidc_silent_refresh_failure(client, mocker):
# mock Keycloak client
mock_keycloak(mocker)
next_path = reverse("swh-web-homepage")
# silent session refresh initialization
login_url = reverse(
"oidc-login", query_params={"next_path": next_path, "prompt": "none"}
)
response = client.get(login_url)
request = response.wsgi_request
login_data = request.session["login_data"]
# check prompt value has been registered in user session
assert "prompt" in login_data
assert login_data["prompt"] == "none"
# simulate a failed silent session refresh
session_state = str(uuid.uuid4())
login_complete_url = reverse(
"oidc-login-complete",
query_params={
"error": "login_required",
"state": login_data["state"],
"session_state": session_state,
},
)
# login process finalization
response = client.get(login_complete_url)
request = response.wsgi_request
# should redirect to logout page
assert response.status_code == 302
logout_url = reverse(
"logout", query_params={"next_path": next_path, "remote_user": 1}
)
assert response["location"] == logout_url
def test_view_rendering_when_user_not_set_in_request(request_factory):
request = request_factory.get("/")
# Django RequestFactory do not set any user by default
assert not hasattr(request, "user")
response = homepage_view(request)
assert response.status_code == 200
diff --git a/swh/web/tests/browse/views/test_content.py b/swh/web/tests/browse/views/test_content.py
index d898f0eb..6af169e0 100644
--- a/swh/web/tests/browse/views/test_content.py
+++ b/swh/web/tests/browse/views/test_content.py
@@ -1,590 +1,591 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
-from django.utils.html import escape
from hypothesis import given
+from django.utils.html import escape
+
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.web.browse.snapshot_context import process_snapshot_branches
from swh.web.browse.utils import (
_re_encode_content,
get_mimetype_and_encoding_for_content,
prepare_content_for_display,
)
from swh.web.common.exc import NotFoundExc
from swh.web.common.identifiers import gen_swhid
from swh.web.common.utils import gen_path_info, reverse
from swh.web.tests.django_asserts import (
assert_contains,
assert_not_contains,
assert_template_used,
)
from swh.web.tests.strategies import (
content,
content_image_type,
content_text,
content_text_no_highlight,
content_text_non_utf8,
content_unsupported_image_type_rendering,
content_utf8_detected_as_binary,
invalid_sha1,
origin_with_multiple_visits,
unknown_content,
)
@given(content_text())
def test_content_view_text(client, archive_data, content):
sha1_git = content["sha1_git"]
url = reverse(
"browse-content",
url_args={"query_string": content["sha1"]},
query_params={"path": content["path"]},
)
url_raw = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display["mimetype"]
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
if mimetype.startswith("text/"):
assert_contains(resp, '<code class="%s">' % content_display["language"])
assert_contains(resp, escape(content_display["content_data"]))
assert_contains(resp, url_raw)
swh_cnt_id = gen_swhid(CONTENT, sha1_git)
swh_cnt_id_url = reverse("browse-swhid", url_args={"swhid": swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
@given(content_text_no_highlight())
def test_content_view_text_no_highlight(client, archive_data, content):
sha1_git = content["sha1_git"]
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
url_raw = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_contains(resp, '<code class="nohighlight">')
assert_contains(resp, escape(content_display["content_data"]))
assert_contains(resp, url_raw)
swh_cnt_id = gen_swhid(CONTENT, sha1_git)
swh_cnt_id_url = reverse("browse-swhid", url_args={"swhid": swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
@given(content_text_non_utf8())
def test_content_view_no_utf8_text(client, archive_data, content):
sha1_git = content["sha1_git"]
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
swh_cnt_id = gen_swhid(CONTENT, sha1_git)
swh_cnt_id_url = reverse("browse-swhid", url_args={"swhid": swh_cnt_id})
assert_contains(resp, swh_cnt_id_url)
assert_contains(resp, escape(content_display["content_data"]))
@given(content_image_type())
def test_content_view_image(client, archive_data, content):
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
url_raw = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display["mimetype"]
content_data = content_display["content_data"]
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_contains(resp, '<img src="data:%s;base64,%s"/>' % (mimetype, content_data))
assert_contains(resp, url_raw)
@given(content_unsupported_image_type_rendering())
def test_content_view_image_no_rendering(client, archive_data, content):
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
resp = client.get(url)
mimetype = content["mimetype"]
encoding = content["encoding"]
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_contains(
resp,
(
f"Content with mime type {mimetype} and encoding {encoding} "
"cannot be displayed."
),
)
@given(content_text())
def test_content_view_text_with_path(client, archive_data, content):
path = content["path"]
url = reverse(
"browse-content",
url_args={"query_string": content["sha1"]},
query_params={"path": path},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_contains(resp, '<nav class="bread-crumbs')
content_display = _process_content_for_display(archive_data, content)
mimetype = content_display["mimetype"]
if mimetype.startswith("text/"):
hljs_language = content["hljs_language"]
assert_contains(resp, '<code class="%s">' % hljs_language)
assert_contains(resp, escape(content_display["content_data"]))
split_path = path.split("/")
root_dir_sha1 = split_path[0]
filename = split_path[-1]
path = path.replace(root_dir_sha1 + "/", "").replace(filename, "")
swhid_context = {
"anchor": gen_swhid(DIRECTORY, root_dir_sha1),
"path": f"/{path}{filename}",
}
swh_cnt_id = gen_swhid(CONTENT, content["sha1_git"], metadata=swhid_context)
swh_cnt_id_url = reverse("browse-swhid", url_args={"swhid": swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
path_info = gen_path_info(path)
root_dir_url = reverse("browse-directory", url_args={"sha1_git": root_dir_sha1})
assert_contains(resp, '<li class="swh-path">', count=len(path_info) + 1)
assert_contains(
resp, '<a href="' + root_dir_url + '">' + root_dir_sha1[:7] + "</a>"
)
for p in path_info:
dir_url = reverse(
"browse-directory",
url_args={"sha1_git": root_dir_sha1},
query_params={"path": p["path"]},
)
assert_contains(resp, '<a href="' + dir_url + '">' + p["name"] + "</a>")
assert_contains(resp, "<li>" + filename + "</li>")
url_raw = reverse(
"browse-content-raw",
url_args={"query_string": content["sha1"]},
query_params={"filename": filename},
)
assert_contains(resp, url_raw)
url = reverse(
"browse-content",
url_args={"query_string": content["sha1"]},
query_params={"path": filename},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_not_contains(resp, '<nav class="bread-crumbs')
invalid_path = "%s/foo/bar/baz" % root_dir_sha1
url = reverse(
"browse-content",
url_args={"query_string": content["sha1"]},
query_params={"path": invalid_path},
)
resp = client.get(url)
assert resp.status_code == 404
@given(content_text())
def test_content_raw_text(client, archive_data, content):
url = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_data = archive_data.content_get_data(content["sha1"])["data"]
assert resp.status_code == 200
assert resp["Content-Type"] == "text/plain"
assert resp["Content-disposition"] == ("filename=%s_%s" % ("sha1", content["sha1"]))
assert resp.content == content_data
filename = content["path"].split("/")[-1]
url = reverse(
"browse-content-raw",
url_args={"query_string": content["sha1"]},
query_params={"filename": filename},
)
resp = client.get(url)
assert resp.status_code == 200
assert resp["Content-Type"] == "text/plain"
assert resp["Content-disposition"] == "filename=%s" % filename
assert resp.content == content_data
@given(content_text_non_utf8())
def test_content_raw_no_utf8_text(client, content):
url = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
assert resp.status_code == 200
_, encoding = get_mimetype_and_encoding_for_content(resp.content)
assert encoding == content["encoding"]
@given(content_image_type())
def test_content_raw_bin(client, archive_data, content):
url = reverse("browse-content-raw", url_args={"query_string": content["sha1"]})
resp = client.get(url)
filename = content["path"].split("/")[-1]
content_data = archive_data.content_get_data(content["sha1"])["data"]
assert resp.status_code == 200
assert resp["Content-Type"] == "application/octet-stream"
assert resp["Content-disposition"] == "attachment; filename=%s_%s" % (
"sha1",
content["sha1"],
)
assert resp.content == content_data
url = reverse(
"browse-content-raw",
url_args={"query_string": content["sha1"]},
query_params={"filename": filename},
)
resp = client.get(url)
assert resp.status_code == 200
assert resp["Content-Type"] == "application/octet-stream"
assert resp["Content-disposition"] == "attachment; filename=%s" % filename
assert resp.content == content_data
@given(invalid_sha1(), unknown_content())
def test_content_request_errors(client, invalid_sha1, unknown_content):
url = reverse("browse-content", url_args={"query_string": invalid_sha1})
resp = client.get(url)
assert resp.status_code == 400
assert_template_used(resp, "error.html")
url = reverse("browse-content", url_args={"query_string": unknown_content["sha1"]})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
@given(content())
def test_content_bytes_missing(client, archive_data, mocker, content):
mock_service = mocker.patch("swh.web.browse.utils.service")
content_data = archive_data.content_get(content["sha1"])
mock_service.lookup_content.return_value = content_data
mock_service.lookup_content_filetype.side_effect = Exception()
mock_service.lookup_content_raw.side_effect = NotFoundExc(
"Content bytes not available!"
)
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "browse/content.html")
def test_content_too_large(client, mocker):
mock_request_content = mocker.patch("swh.web.browse.views.content.request_content")
stub_content_too_large_data = {
"checksums": {
"sha1": "8624bcdae55baeef00cd11d5dfcfa60f68710a02",
"sha1_git": "94a9ed024d3859793618152ea559a168bbcbb5e2",
"sha256": (
"8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b16" "5a1dcd80c7c545eb65b903"
),
"blake2s256": (
"38702b7168c7785bfe748b51b45d9856070ba90" "f9dc6d90f2ea75d4356411ffe"
),
},
"length": 30000000,
"raw_data": None,
"mimetype": "text/plain",
"encoding": "us-ascii",
"language": "not detected",
"licenses": "GPL",
"error_code": 200,
"error_message": "",
"error_description": "",
}
content_sha1 = stub_content_too_large_data["checksums"]["sha1"]
mock_request_content.return_value = stub_content_too_large_data
url = reverse("browse-content", url_args={"query_string": content_sha1})
url_raw = reverse("browse-content-raw", url_args={"query_string": content_sha1})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert_contains(resp, "Content is too large to be displayed")
assert_contains(resp, url_raw)
@given(content())
def test_content_uppercase(client, content):
url = reverse(
"browse-content-uppercase-checksum",
url_args={"query_string": content["sha1"].upper()},
)
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse("browse-content", url_args={"query_string": content["sha1"]})
assert resp["location"] == redirect_url
@given(content_utf8_detected_as_binary())
def test_content_utf8_detected_as_binary_display(client, archive_data, content):
url = reverse("browse-content", url_args={"query_string": content["sha1"]})
resp = client.get(url)
content_display = _process_content_for_display(archive_data, content)
assert_contains(resp, escape(content_display["content_data"]))
@given(origin_with_multiple_visits())
def test_content_origin_snapshot_branch_browse(client, archive_data, origin):
visits = archive_data.origin_visit_get(origin["url"])
visit = random.choice(visits)
snapshot = archive_data.snapshot_get(visit["snapshot"])
branches, releases = process_snapshot_branches(snapshot)
branch_info = random.choice(branches)
directory = archive_data.revision_get(branch_info["revision"])["directory"]
directory_content = archive_data.directory_ls(directory)
directory_file = random.choice(
[e for e in directory_content if e["type"] == "file"]
)
url = reverse(
"browse-content",
url_args={"query_string": directory_file["checksums"]["sha1"]},
query_params={
"origin_url": origin["url"],
"snapshot": snapshot["id"],
"branch": branch_info["name"],
"path": directory_file["name"],
},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
_check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases)
assert_contains(resp, directory_file["name"])
assert_contains(resp, f"Branch: <strong>{branch_info['name']}</strong>")
cnt_swhid = gen_swhid(
CONTENT,
directory_file["checksums"]["sha1_git"],
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(REVISION, branch_info["revision"]),
"path": f"/{directory_file['name']}",
},
)
assert_contains(resp, cnt_swhid)
dir_swhid = gen_swhid(
DIRECTORY,
directory,
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(REVISION, branch_info["revision"]),
"path": "/",
},
)
assert_contains(resp, dir_swhid)
rev_swhid = gen_swhid(
REVISION,
branch_info["revision"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rev_swhid)
snp_swhid = gen_swhid(SNAPSHOT, snapshot, metadata={"origin": origin["url"],},)
assert_contains(resp, snp_swhid)
@given(origin_with_multiple_visits())
def test_content_origin_snapshot_release_browse(client, archive_data, origin):
visits = archive_data.origin_visit_get(origin["url"])
visit = random.choice(visits)
snapshot = archive_data.snapshot_get(visit["snapshot"])
branches, releases = process_snapshot_branches(snapshot)
release_info = random.choice(releases)
directory_content = archive_data.directory_ls(release_info["directory"])
directory_file = random.choice(
[e for e in directory_content if e["type"] == "file"]
)
url = reverse(
"browse-content",
url_args={"query_string": directory_file["checksums"]["sha1"]},
query_params={
"origin_url": origin["url"],
"snapshot": snapshot["id"],
"release": release_info["name"],
"path": directory_file["name"],
},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
_check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases)
assert_contains(resp, directory_file["name"])
assert_contains(resp, f"Release: <strong>{release_info['name']}</strong>")
cnt_swhid = gen_swhid(
CONTENT,
directory_file["checksums"]["sha1_git"],
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(RELEASE, release_info["id"]),
"path": f"/{directory_file['name']}",
},
)
assert_contains(resp, cnt_swhid)
dir_swhid = gen_swhid(
DIRECTORY,
release_info["directory"],
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(RELEASE, release_info["id"]),
"path": "/",
},
)
assert_contains(resp, dir_swhid)
rev_swhid = gen_swhid(
REVISION,
release_info["target"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rev_swhid)
rel_swhid = gen_swhid(
RELEASE,
release_info["id"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rel_swhid)
snp_swhid = gen_swhid(SNAPSHOT, snapshot, metadata={"origin": origin["url"],},)
assert_contains(resp, snp_swhid)
def _check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases):
browse_origin_url = reverse(
"browse-origin", query_params={"origin_url": origin["url"]}
)
assert_contains(resp, f'href="{browse_origin_url}"')
origin_branches_url = reverse(
"browse-origin-branches",
query_params={"origin_url": origin["url"], "snapshot": snapshot["id"]},
)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(branches)})")
origin_releases_url = reverse(
"browse-origin-releases",
query_params={"origin_url": origin["url"], "snapshot": snapshot["id"]},
)
assert_contains(resp, f'href="{escape(origin_releases_url)}"')
assert_contains(resp, f"Releases ({len(releases)})")
assert_contains(resp, '<li class="swh-branch">', count=len(branches))
assert_contains(resp, '<li class="swh-release">', count=len(releases))
def _process_content_for_display(archive_data, content):
content_data = archive_data.content_get_data(content["sha1"])
mime_type, encoding = get_mimetype_and_encoding_for_content(content_data["data"])
mime_type, encoding, content_data = _re_encode_content(
mime_type, encoding, content_data["data"]
)
content_display = prepare_content_for_display(
content_data, mime_type, content["path"]
)
assert type(content_display["content_data"]) == str
return content_display
diff --git a/swh/web/tests/browse/views/test_directory.py b/swh/web/tests/browse/views/test_directory.py
index 57479ad0..8b3e314e 100644
--- a/swh/web/tests/browse/views/test_directory.py
+++ b/swh/web/tests/browse/views/test_directory.py
@@ -1,317 +1,318 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
-from django.utils.html import escape
from hypothesis import given
+from django.utils.html import escape
+
from swh.model.identifiers import DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.web.browse.snapshot_context import process_snapshot_branches
from swh.web.common.identifiers import gen_swhid
from swh.web.common.utils import gen_path_info, reverse
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.tests.strategies import (
directory,
directory_with_subdirs,
invalid_sha1,
origin_with_multiple_visits,
unknown_directory,
)
@given(directory())
def test_root_directory_view(client, archive_data, directory):
_directory_view_checks(client, directory, archive_data.directory_ls(directory))
@given(directory_with_subdirs())
def test_sub_directory_view(client, archive_data, directory):
dir_content = archive_data.directory_ls(directory)
subdir = random.choice([e for e in dir_content if e["type"] == "dir"])
subdir_content = archive_data.directory_ls(subdir["target"])
_directory_view_checks(client, directory, subdir_content, subdir["name"])
@given(invalid_sha1(), unknown_directory())
def test_directory_request_errors(client, invalid_sha1, unknown_directory):
dir_url = reverse("browse-directory", url_args={"sha1_git": invalid_sha1})
resp = client.get(dir_url)
assert resp.status_code == 400
assert_template_used(resp, "error.html")
dir_url = reverse("browse-directory", url_args={"sha1_git": unknown_directory})
resp = client.get(dir_url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
@given(directory())
def test_directory_uppercase(client, directory):
url = reverse(
"browse-directory-uppercase-checksum", url_args={"sha1_git": directory.upper()}
)
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse("browse-directory", url_args={"sha1_git": directory})
assert resp["location"] == redirect_url
@given(directory())
def test_permalink_box_context(client, tests_data, directory):
origin_url = random.choice(tests_data["origins"])["url"]
url = reverse(
"browse-directory",
url_args={"sha1_git": directory},
query_params={"origin_url": origin_url},
)
resp = client.get(url)
assert resp.status_code == 200
assert_contains(resp, 'id="swhid-context-option-directory"')
@given(origin_with_multiple_visits())
def test_directory_origin_snapshot_branch_browse(client, archive_data, origin):
visits = archive_data.origin_visit_get(origin["url"])
visit = random.choice(visits)
snapshot = archive_data.snapshot_get(visit["snapshot"])
branches, releases = process_snapshot_branches(snapshot)
branch_info = random.choice(branches)
directory = archive_data.revision_get(branch_info["revision"])["directory"]
directory_content = archive_data.directory_ls(directory)
directory_subdir = random.choice(
[e for e in directory_content if e["type"] == "dir"]
)
url = reverse(
"browse-directory",
url_args={"sha1_git": directory},
query_params={
"origin_url": origin["url"],
"snapshot": snapshot["id"],
"branch": branch_info["name"],
"path": directory_subdir["name"],
},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
_check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases)
assert_contains(resp, directory_subdir["name"])
assert_contains(resp, f"Branch: <strong>{branch_info['name']}</strong>")
dir_swhid = gen_swhid(
DIRECTORY,
directory_subdir["target"],
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(REVISION, branch_info["revision"]),
"path": "/",
},
)
assert_contains(resp, dir_swhid)
rev_swhid = gen_swhid(
REVISION,
branch_info["revision"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rev_swhid)
snp_swhid = gen_swhid(SNAPSHOT, snapshot, metadata={"origin": origin["url"],},)
assert_contains(resp, snp_swhid)
@given(origin_with_multiple_visits())
def test_content_origin_snapshot_release_browse(client, archive_data, origin):
visits = archive_data.origin_visit_get(origin["url"])
visit = random.choice(visits)
snapshot = archive_data.snapshot_get(visit["snapshot"])
branches, releases = process_snapshot_branches(snapshot)
release_info = random.choice(releases)
directory = release_info["directory"]
directory_content = archive_data.directory_ls(directory)
directory_subdir = random.choice(
[e for e in directory_content if e["type"] == "dir"]
)
url = reverse(
"browse-directory",
url_args={"sha1_git": directory},
query_params={
"origin_url": origin["url"],
"snapshot": snapshot["id"],
"release": release_info["name"],
"path": directory_subdir["name"],
},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
_check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases)
assert_contains(resp, directory_subdir["name"])
assert_contains(resp, f"Release: <strong>{release_info['name']}</strong>")
dir_swhid = gen_swhid(
DIRECTORY,
directory_subdir["target"],
metadata={
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot),
"anchor": gen_swhid(RELEASE, release_info["id"]),
"path": "/",
},
)
assert_contains(resp, dir_swhid)
rev_swhid = gen_swhid(
REVISION,
release_info["target"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rev_swhid)
rel_swhid = gen_swhid(
RELEASE,
release_info["id"],
metadata={"origin": origin["url"], "visit": gen_swhid(SNAPSHOT, snapshot),},
)
assert_contains(resp, rel_swhid)
snp_swhid = gen_swhid(SNAPSHOT, snapshot, metadata={"origin": origin["url"],},)
assert_contains(resp, snp_swhid)
def _check_origin_snapshot_related_html(resp, origin, snapshot, branches, releases):
browse_origin_url = reverse(
"browse-origin", query_params={"origin_url": origin["url"]}
)
assert_contains(resp, f'href="{browse_origin_url}"')
origin_branches_url = reverse(
"browse-origin-branches",
query_params={"origin_url": origin["url"], "snapshot": snapshot["id"]},
)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(branches)})")
origin_releases_url = reverse(
"browse-origin-releases",
query_params={"origin_url": origin["url"], "snapshot": snapshot["id"]},
)
assert_contains(resp, f'href="{escape(origin_releases_url)}"')
assert_contains(resp, f"Releases ({len(releases)})")
assert_contains(resp, '<li class="swh-branch">', count=len(branches))
assert_contains(resp, '<li class="swh-release">', count=len(releases))
def _directory_view_checks(
client,
root_directory_sha1,
directory_entries,
path=None,
origin_url=None,
snapshot_id=None,
):
dirs = [e for e in directory_entries if e["type"] in ("dir", "rev")]
files = [e for e in directory_entries if e["type"] == "file"]
url_args = {"sha1_git": root_directory_sha1}
query_params = {"path": path, "origin_url": origin_url, "snapshot": snapshot_id}
url = reverse("browse-directory", url_args=url_args, query_params=query_params)
root_dir_url = reverse(
"browse-directory", url_args={"sha1_git": root_directory_sha1}
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
assert_contains(
resp, '<a href="' + root_dir_url + '">' + root_directory_sha1[:7] + "</a>"
)
assert_contains(resp, '<td class="swh-directory">', count=len(dirs))
assert_contains(resp, '<td class="swh-content">', count=len(files))
for d in dirs:
if d["type"] == "rev":
dir_url = reverse("browse-revision", url_args={"sha1_git": d["target"]})
else:
dir_path = d["name"]
if path:
dir_path = "%s/%s" % (path, d["name"])
dir_url = reverse(
"browse-directory",
url_args={"sha1_git": root_directory_sha1},
query_params={"path": dir_path},
)
assert_contains(resp, dir_url)
for f in files:
file_path = "%s/%s" % (root_directory_sha1, f["name"])
if path:
file_path = "%s/%s/%s" % (root_directory_sha1, path, f["name"])
query_string = "sha1_git:" + f["target"]
file_url = reverse(
"browse-content",
url_args={"query_string": query_string},
query_params={"path": file_path},
)
assert_contains(resp, file_url)
path_info = gen_path_info(path)
assert_contains(resp, '<li class="swh-path">', count=len(path_info) + 1)
assert_contains(
resp, '<a href="%s">%s</a>' % (root_dir_url, root_directory_sha1[:7])
)
for p in path_info:
dir_url = reverse(
"browse-directory",
url_args={"sha1_git": root_directory_sha1},
query_params={"path": p["path"]},
)
assert_contains(resp, '<a href="%s">%s</a>' % (dir_url, p["name"]))
assert_contains(resp, "vault-cook-directory")
swh_dir_id = gen_swhid(DIRECTORY, directory_entries[0]["dir_id"])
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
swhid_context = {}
if root_directory_sha1 != directory_entries[0]["dir_id"]:
swhid_context["anchor"] = gen_swhid(DIRECTORY, root_directory_sha1)
swhid_context["path"] = f"/{path}/" if path else "/"
if root_directory_sha1 != directory_entries[0]["dir_id"]:
swhid_context["anchor"] = gen_swhid(DIRECTORY, root_directory_sha1)
swh_dir_id = gen_swhid(
DIRECTORY, directory_entries[0]["dir_id"], metadata=swhid_context
)
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
diff --git a/swh/web/tests/browse/views/test_origin.py b/swh/web/tests/browse/views/test_origin.py
index 1ad16dfe..be4ad580 100644
--- a/swh/web/tests/browse/views/test_origin.py
+++ b/swh/web/tests/browse/views/test_origin.py
@@ -1,1244 +1,1245 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
import re
import string
-from django.utils.html import escape
from hypothesis import given
+from django.utils.html import escape
+
from swh.model.hashutil import hash_to_bytes
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.model.model import (
OriginVisit,
OriginVisitStatus,
Snapshot,
SnapshotBranch,
TargetType,
)
from swh.storage.utils import now
from swh.web.browse.snapshot_context import process_snapshot_branches
from swh.web.common.exc import NotFoundExc
from swh.web.common.identifiers import gen_swhid
from swh.web.common.utils import (
format_utc_iso_date,
gen_path_info,
parse_iso8601_date_to_utc,
reverse,
)
from swh.web.tests.data import get_content, random_sha1
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.tests.strategies import (
new_origin,
new_snapshot,
origin,
origin_with_multiple_visits,
origin_with_releases,
)
from swh.web.tests.strategies import release as existing_release
from swh.web.tests.strategies import revisions, unknown_revision, visit_dates
@given(origin_with_multiple_visits())
def test_origin_visits_browse(client, archive_data, origin):
url = reverse("browse-origin-visits", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/origin-visits.html")
url = reverse("browse-origin-visits", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/origin-visits.html")
visits = archive_data.origin_visit_get(origin["url"])
for v in visits:
vdate = format_utc_iso_date(v["date"], "%Y-%m-%dT%H:%M:%SZ")
browse_dir_url = reverse(
"browse-origin-directory",
query_params={"origin_url": origin["url"], "timestamp": vdate},
)
assert_contains(resp, browse_dir_url)
_check_origin_link(resp, origin["url"])
@given(origin_with_multiple_visits())
def test_origin_content_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin["url"])
def _get_archive_data(visit_idx):
snapshot = archive_data.snapshot_get(origin_visits[visit_idx]["snapshot"])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
dir_content = archive_data.directory_ls(head_rev["directory"])
dir_files = [e for e in dir_content if e["type"] == "file"]
dir_file = random.choice(dir_files)
branches, releases = process_snapshot_branches(snapshot)
return {
"branches": branches,
"releases": releases,
"root_dir_sha1": head_rev["directory"],
"content": get_content(dir_file["checksums"]["sha1"]),
"visit": origin_visits[visit_idx],
}
tdata = _get_archive_data(-1)
_origin_content_view_test_helper(
client,
archive_data,
origin,
origin_visits[-1],
tdata["branches"],
tdata["releases"],
tdata["root_dir_sha1"],
tdata["content"],
)
_origin_content_view_test_helper(
client,
archive_data,
origin,
origin_visits[-1],
tdata["branches"],
tdata["releases"],
tdata["root_dir_sha1"],
tdata["content"],
timestamp=tdata["visit"]["date"],
)
_origin_content_view_test_helper(
client,
archive_data,
origin,
origin_visits[-1],
tdata["branches"],
tdata["releases"],
tdata["root_dir_sha1"],
tdata["content"],
snapshot_id=tdata["visit"]["snapshot"],
)
tdata = _get_archive_data(0)
_origin_content_view_test_helper(
client,
archive_data,
origin,
origin_visits[0],
tdata["branches"],
tdata["releases"],
tdata["root_dir_sha1"],
tdata["content"],
visit_id=tdata["visit"]["visit"],
)
_origin_content_view_test_helper(
client,
archive_data,
origin,
origin_visits[0],
tdata["branches"],
tdata["releases"],
tdata["root_dir_sha1"],
tdata["content"],
snapshot_id=tdata["visit"]["snapshot"],
)
@given(origin())
def test_origin_root_directory_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin["url"])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit["snapshot"])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
root_dir_sha1 = head_rev["directory"]
dir_content = archive_data.directory_ls(root_dir_sha1)
branches, releases = process_snapshot_branches(snapshot)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
visit_id=visit["visit"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
timestamp=visit["date"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
snapshot_id=visit["snapshot"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
visit_id=visit["visit"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
timestamp=visit["date"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
dir_content,
snapshot_id=visit["snapshot"],
)
@given(origin())
def test_origin_sub_directory_view(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin["url"])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit["snapshot"])
head_rev_id = archive_data.snapshot_get_head(snapshot)
head_rev = archive_data.revision_get(head_rev_id)
root_dir_sha1 = head_rev["directory"]
subdirs = [
e for e in archive_data.directory_ls(root_dir_sha1) if e["type"] == "dir"
]
branches, releases = process_snapshot_branches(snapshot)
if len(subdirs) == 0:
return
subdir = random.choice(subdirs)
subdir_content = archive_data.directory_ls(subdir["target"])
subdir_path = subdir["name"]
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
visit_id=visit["visit"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
timestamp=visit["date"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
snapshot_id=visit["snapshot"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
visit_id=visit["visit"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
timestamp=visit["date"],
)
_origin_directory_view_test_helper(
client,
archive_data,
origin,
visit,
branches,
releases,
root_dir_sha1,
subdir_content,
path=subdir_path,
snapshot_id=visit["snapshot"],
)
@given(origin())
def test_origin_branches(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin["url"])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit["snapshot"])
snapshot_content = process_snapshot_branches(snapshot)
_origin_branches_test_helper(client, origin, snapshot_content)
_origin_branches_test_helper(
client, origin, snapshot_content, snapshot_id=visit["snapshot"]
)
@given(origin())
def test_origin_releases(client, archive_data, origin):
origin_visits = archive_data.origin_visit_get(origin["url"])
visit = origin_visits[-1]
snapshot = archive_data.snapshot_get(visit["snapshot"])
snapshot_content = process_snapshot_branches(snapshot)
_origin_releases_test_helper(client, origin, snapshot_content)
_origin_releases_test_helper(
client, origin, snapshot_content, snapshot_id=visit["snapshot"]
)
@given(
new_origin(),
new_snapshot(min_size=4, max_size=4),
visit_dates(),
revisions(min_size=3, max_size=3),
)
def test_origin_snapshot_null_branch(
client, archive_data, new_origin, new_snapshot, visit_dates, revisions
):
snp_dict = new_snapshot.to_dict()
archive_data.origin_add([new_origin])
for i, branch in enumerate(snp_dict["branches"].keys()):
if i == 0:
snp_dict["branches"][branch] = None
else:
snp_dict["branches"][branch] = {
"target_type": "revision",
"target": hash_to_bytes(revisions[i - 1]),
}
archive_data.snapshot_add([Snapshot.from_dict(snp_dict)])
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_dates[0], type="git",)]
)[0]
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=now(),
status="partial",
snapshot=snp_dict["id"],
)
archive_data.origin_visit_status_add([visit_status])
url = reverse(
"browse-origin-directory", query_params={"origin_url": new_origin.url}
)
rv = client.get(url)
assert rv.status_code == 200
@given(
new_origin(),
new_snapshot(min_size=4, max_size=4),
visit_dates(),
revisions(min_size=4, max_size=4),
)
def test_origin_snapshot_invalid_branch(
client, archive_data, new_origin, new_snapshot, visit_dates, revisions
):
snp_dict = new_snapshot.to_dict()
archive_data.origin_add([new_origin])
for i, branch in enumerate(snp_dict["branches"].keys()):
snp_dict["branches"][branch] = {
"target_type": "revision",
"target": hash_to_bytes(revisions[i]),
}
archive_data.snapshot_add([Snapshot.from_dict(snp_dict)])
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_dates[0], type="git",)]
)[0]
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=now(),
status="full",
snapshot=snp_dict["id"],
)
archive_data.origin_visit_status_add([visit_status])
url = reverse(
"browse-origin-directory",
query_params={"origin_url": new_origin.url, "branch": "invalid_branch"},
)
rv = client.get(url)
assert rv.status_code == 404
@given(new_origin())
def test_browse_visits_origin_not_found(client, new_origin):
url = reverse("browse-origin-visits", query_params={"origin_url": new_origin.url})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(
resp, f"Origin with url {new_origin.url} not found", status_code=404
)
@given(origin())
def test_browse_origin_directory_no_visit(client, mocker, origin):
mock_get_origin_visits = mocker.patch(
"swh.web.common.origin_visits.get_origin_visits"
)
mock_get_origin_visits.return_value = []
mock_service = mocker.patch("swh.web.common.origin_visits.service")
mock_service.lookup_origin_visit_latest.return_value = None
url = reverse("browse-origin-directory", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(resp, "No valid visit", status_code=404)
assert not mock_get_origin_visits.called
@given(origin())
def test_browse_origin_directory_unknown_visit(client, mocker, origin):
mock_get_origin_visits = mocker.patch(
"swh.web.common.origin_visits.get_origin_visits"
)
mock_get_origin_visits.return_value = [{"visit": 1}]
url = reverse(
"browse-origin-directory",
query_params={"origin_url": origin["url"], "visit_id": 2},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert re.search("Visit.*not found", resp.content.decode("utf-8"))
assert mock_get_origin_visits.called
@given(origin())
def test_browse_origin_directory_not_found(client, origin):
url = reverse(
"browse-origin-directory",
query_params={"origin_url": origin["url"], "path": "/invalid/dir/path/"},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert re.search("Directory.*not found", resp.content.decode("utf-8"))
@given(origin())
def test_browse_origin_content_no_visit(client, mocker, origin):
mock_get_origin_visits = mocker.patch(
"swh.web.common.origin_visits.get_origin_visits"
)
mock_get_origin_visits.return_value = []
mock_service = mocker.patch("swh.web.common.origin_visits.service")
mock_service.lookup_origin_visit_latest.return_value = None
url = reverse(
"browse-origin-content",
query_params={"origin_url": origin["url"], "path": "foo"},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(resp, "No valid visit", status_code=404)
assert not mock_get_origin_visits.called
@given(origin())
def test_browse_origin_content_unknown_visit(client, mocker, origin):
mock_get_origin_visits = mocker.patch(
"swh.web.common.origin_visits.get_origin_visits"
)
mock_get_origin_visits.return_value = [{"visit": 1}]
url = reverse(
"browse-origin-content",
query_params={"origin_url": origin["url"], "path": "foo", "visit_id": 2},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert re.search("Visit.*not found", resp.content.decode("utf-8"))
assert mock_get_origin_visits.called
@given(origin())
def test_browse_origin_content_directory_empty_snapshot(client, mocker, origin):
mock_snapshot_service = mocker.patch("swh.web.browse.snapshot_context.service")
mock_get_origin_visit_snapshot = mocker.patch(
"swh.web.browse.snapshot_context.get_origin_visit_snapshot"
)
mock_get_origin_visit_snapshot.return_value = ([], [])
mock_snapshot_service.lookup_origin.return_value = origin
mock_snapshot_service.lookup_snapshot_sizes.return_value = {
"revision": 0,
"release": 0,
}
for browse_context in ("content", "directory"):
url = reverse(
f"browse-origin-{browse_context}",
query_params={"origin_url": origin["url"], "path": "baz"},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, f"browse/{browse_context}.html")
assert re.search("snapshot.*is empty", resp.content.decode("utf-8"))
assert mock_get_origin_visit_snapshot.called
assert mock_snapshot_service.lookup_origin.called
assert mock_snapshot_service.lookup_snapshot_sizes.called
@given(origin())
def test_browse_origin_content_not_found(client, origin):
url = reverse(
"browse-origin-content",
query_params={"origin_url": origin["url"], "path": "/invalid/file/path"},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert re.search("Directory entry.*not found", resp.content.decode("utf-8"))
@given(origin())
def test_browse_directory_snapshot_not_found(client, mocker, origin):
mock_get_snapshot_context = mocker.patch(
"swh.web.browse.snapshot_context.get_snapshot_context"
)
mock_get_snapshot_context.side_effect = NotFoundExc("Snapshot not found")
url = reverse("browse-origin-directory", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(resp, "Snapshot not found", status_code=404)
assert mock_get_snapshot_context.called
@given(origin())
def test_origin_empty_snapshot(client, mocker, origin):
mock_service = mocker.patch("swh.web.browse.snapshot_context.service")
mock_get_origin_visit_snapshot = mocker.patch(
"swh.web.browse.snapshot_context.get_origin_visit_snapshot"
)
mock_get_origin_visit_snapshot.return_value = ([], [])
mock_service.lookup_snapshot_sizes.return_value = {
"revision": 0,
"release": 0,
}
mock_service.lookup_origin.return_value = origin
url = reverse("browse-origin-directory", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
resp_content = resp.content.decode("utf-8")
assert re.search("snapshot.*is empty", resp_content)
assert not re.search("swh-tr-link", resp_content)
assert mock_get_origin_visit_snapshot.called
assert mock_service.lookup_snapshot_sizes.called
@given(new_origin())
def test_origin_empty_snapshot_null_revision(client, archive_data, new_origin):
snapshot = Snapshot(
branches={
b"HEAD": SnapshotBranch(
target="refs/head/master".encode(), target_type=TargetType.ALIAS,
),
b"refs/head/master": None,
}
)
archive_data.origin_add([new_origin])
archive_data.snapshot_add([snapshot])
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=now(), type="git",)]
)[0]
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=now(),
status="partial",
snapshot=snapshot.id,
)
archive_data.origin_visit_status_add([visit_status])
url = reverse(
"browse-origin-directory", query_params={"origin_url": new_origin.url},
)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
resp_content = resp.content.decode("utf-8")
assert re.search("snapshot.*is empty", resp_content)
assert not re.search("swh-tr-link", resp_content)
@given(origin_with_releases())
def test_origin_release_browse(client, archive_data, origin):
snapshot = archive_data.snapshot_get_latest(origin["url"])
release = [
b for b in snapshot["branches"].values() if b["target_type"] == "release"
][-1]
release_data = archive_data.release_get(release["target"])
revision_data = archive_data.revision_get(release_data["target"])
url = reverse(
"browse-origin-directory",
query_params={"origin_url": origin["url"], "release": release_data["name"]},
)
resp = client.get(url)
assert resp.status_code == 200
assert_contains(resp, release_data["name"])
assert_contains(resp, release["target"])
swhid_context = {
"origin": origin["url"],
"visit": gen_swhid(SNAPSHOT, snapshot["id"]),
"anchor": gen_swhid(RELEASE, release_data["id"]),
"path": "/",
}
swh_dir_id = gen_swhid(
DIRECTORY, revision_data["directory"], metadata=swhid_context
)
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
@given(origin_with_releases())
def test_origin_release_browse_not_found(client, origin):
invalid_release_name = "swh-foo-bar"
url = reverse(
"browse-origin-directory",
query_params={"origin_url": origin["url"], "release": invalid_release_name},
)
resp = client.get(url)
assert resp.status_code == 404
assert re.search(
f"Release {invalid_release_name}.*not found", resp.content.decode("utf-8")
)
@given(new_origin(), unknown_revision())
def test_origin_browse_directory_branch_with_non_resolvable_revision(
client, archive_data, new_origin, unknown_revision
):
branch_name = "master"
snapshot = Snapshot(
branches={
branch_name.encode(): SnapshotBranch(
target=hash_to_bytes(unknown_revision), target_type=TargetType.REVISION,
)
}
)
archive_data.origin_add([new_origin])
archive_data.snapshot_add([snapshot])
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=now(), type="git",)]
)[0]
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=now(),
status="partial",
snapshot=snapshot.id,
)
archive_data.origin_visit_status_add([visit_status])
url = reverse(
"browse-origin-directory",
query_params={"origin_url": new_origin.url, "branch": branch_name},
)
resp = client.get(url)
assert resp.status_code == 200
assert_contains(
resp, f"Revision {unknown_revision } could not be found in the archive."
)
@given(origin())
def test_origin_content_no_path(client, origin):
url = reverse("browse-origin-content", query_params={"origin_url": origin["url"]})
resp = client.get(url)
assert resp.status_code == 400
assert_contains(
resp, "The path of a content must be given as query parameter.", status_code=400
)
def test_origin_views_no_url_query_parameter(client):
for browse_context in (
"content",
"directory",
"log",
"branches",
"releases",
"visits",
):
url = reverse(f"browse-origin-{browse_context}")
resp = client.get(url)
assert resp.status_code == 400
assert_contains(
resp, "An origin URL must be provided as query parameter.", status_code=400
)
def _origin_content_view_test_helper(
client,
archive_data,
origin_info,
origin_visit,
origin_branches,
origin_releases,
root_dir_sha1,
content,
visit_id=None,
timestamp=None,
snapshot_id=None,
):
content_path = "/".join(content["path"].split("/")[1:])
if not visit_id and not snapshot_id:
visit_id = origin_visit["visit"]
query_params = {"origin_url": origin_info["url"], "path": content_path}
if timestamp:
query_params["timestamp"] = timestamp
if visit_id:
query_params["visit_id"] = visit_id
elif snapshot_id:
query_params["snapshot"] = snapshot_id
url = reverse("browse-origin-content", query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
assert type(content["data"]) == str
assert_contains(resp, '<code class="%s">' % content["hljs_language"])
assert_contains(resp, escape(content["data"]))
split_path = content_path.split("/")
filename = split_path[-1]
path = content_path.replace(filename, "")[:-1]
path_info = gen_path_info(path)
del query_params["path"]
if timestamp:
query_params["timestamp"] = format_utc_iso_date(
parse_iso8601_date_to_utc(timestamp).isoformat(), "%Y-%m-%dT%H:%M:%SZ"
)
root_dir_url = reverse("browse-origin-directory", query_params=query_params)
assert_contains(resp, '<li class="swh-path">', count=len(path_info) + 1)
assert_contains(resp, '<a href="%s">%s</a>' % (root_dir_url, root_dir_sha1[:7]))
for p in path_info:
query_params["path"] = p["path"]
dir_url = reverse("browse-origin-directory", query_params=query_params)
assert_contains(resp, '<a href="%s">%s</a>' % (dir_url, p["name"]))
assert_contains(resp, "<li>%s</li>" % filename)
query_string = "sha1_git:" + content["sha1_git"]
url_raw = reverse(
"browse-content-raw",
url_args={"query_string": query_string},
query_params={"filename": filename},
)
assert_contains(resp, url_raw)
if "path" in query_params:
del query_params["path"]
origin_branches_url = reverse("browse-origin-branches", query_params=query_params)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(origin_branches)})")
origin_releases_url = reverse("browse-origin-releases", query_params=query_params)
assert_contains(resp, f'href="{escape(origin_releases_url)}">')
assert_contains(resp, f"Releases ({len(origin_releases)})")
assert_contains(resp, '<li class="swh-branch">', count=len(origin_branches))
query_params["path"] = content_path
for branch in origin_branches:
root_dir_branch_url = reverse(
"browse-origin-content",
query_params={"branch": branch["name"], **query_params},
)
assert_contains(resp, '<a href="%s">' % root_dir_branch_url)
assert_contains(resp, '<li class="swh-release">', count=len(origin_releases))
query_params["branch"] = None
for release in origin_releases:
root_dir_release_url = reverse(
"browse-origin-content",
query_params={"release": release["name"], **query_params},
)
assert_contains(resp, '<a href="%s">' % root_dir_release_url)
url = reverse("browse-origin-content", query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/content.html")
snapshot = archive_data.snapshot_get(origin_visit["snapshot"])
head_rev_id = archive_data.snapshot_get_head(snapshot)
swhid_context = {
"origin": origin_info["url"],
"visit": gen_swhid(SNAPSHOT, snapshot["id"]),
"anchor": gen_swhid(REVISION, head_rev_id),
"path": f"/{content_path}",
}
swh_cnt_id = gen_swhid(CONTENT, content["sha1_git"], metadata=swhid_context)
swh_cnt_id_url = reverse("browse-swhid", url_args={"swhid": swh_cnt_id})
assert_contains(resp, swh_cnt_id)
assert_contains(resp, swh_cnt_id_url)
assert_contains(resp, "swh-take-new-snapshot")
_check_origin_link(resp, origin_info["url"])
def _origin_directory_view_test_helper(
client,
archive_data,
origin_info,
origin_visit,
origin_branches,
origin_releases,
root_directory_sha1,
directory_entries,
visit_id=None,
timestamp=None,
snapshot_id=None,
path=None,
):
dirs = [e for e in directory_entries if e["type"] in ("dir", "rev")]
files = [e for e in directory_entries if e["type"] == "file"]
if not visit_id and not snapshot_id:
visit_id = origin_visit["visit"]
query_params = {"origin_url": origin_info["url"]}
if timestamp:
query_params["timestamp"] = timestamp
elif visit_id:
query_params["visit_id"] = visit_id
else:
query_params["snapshot"] = snapshot_id
if path:
query_params["path"] = path
url = reverse("browse-origin-directory", query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
assert resp.status_code == 200
assert_template_used(resp, "browse/directory.html")
assert_contains(resp, '<td class="swh-directory">', count=len(dirs))
assert_contains(resp, '<td class="swh-content">', count=len(files))
if timestamp:
query_params["timestamp"] = format_utc_iso_date(
parse_iso8601_date_to_utc(timestamp).isoformat(), "%Y-%m-%dT%H:%M:%SZ"
)
for d in dirs:
if d["type"] == "rev":
dir_url = reverse("browse-revision", url_args={"sha1_git": d["target"]})
else:
dir_path = d["name"]
if path:
dir_path = "%s/%s" % (path, d["name"])
query_params["path"] = dir_path
dir_url = reverse("browse-origin-directory", query_params=query_params,)
assert_contains(resp, dir_url)
for f in files:
file_path = f["name"]
if path:
file_path = "%s/%s" % (path, f["name"])
query_params["path"] = file_path
file_url = reverse("browse-origin-content", query_params=query_params)
assert_contains(resp, file_url)
if "path" in query_params:
del query_params["path"]
root_dir_branch_url = reverse("browse-origin-directory", query_params=query_params)
nb_bc_paths = 1
if path:
nb_bc_paths = len(path.split("/")) + 1
assert_contains(resp, '<li class="swh-path">', count=nb_bc_paths)
assert_contains(
resp, '<a href="%s">%s</a>' % (root_dir_branch_url, root_directory_sha1[:7])
)
origin_branches_url = reverse("browse-origin-branches", query_params=query_params)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(origin_branches)})")
origin_releases_url = reverse("browse-origin-releases", query_params=query_params)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, f'href="{escape(origin_releases_url)}"')
assert_contains(resp, f"Releases ({nb_releases})")
if path:
query_params["path"] = path
assert_contains(resp, '<li class="swh-branch">', count=len(origin_branches))
for branch in origin_branches:
query_params["branch"] = branch["name"]
root_dir_branch_url = reverse(
"browse-origin-directory", query_params=query_params
)
assert_contains(resp, '<a href="%s">' % root_dir_branch_url)
assert_contains(resp, '<li class="swh-release">', count=len(origin_releases))
query_params["branch"] = None
for release in origin_releases:
query_params["release"] = release["name"]
root_dir_release_url = reverse(
"browse-origin-directory", query_params=query_params
)
assert_contains(resp, 'href="%s"' % root_dir_release_url)
assert_contains(resp, "vault-cook-directory")
assert_contains(resp, "vault-cook-revision")
snapshot = archive_data.snapshot_get(origin_visit["snapshot"])
head_rev_id = archive_data.snapshot_get_head(snapshot)
swhid_context = {
"origin": origin_info["url"],
"visit": gen_swhid(SNAPSHOT, snapshot["id"]),
"anchor": gen_swhid(REVISION, head_rev_id),
"path": f"/{path}" if path else "/",
}
swh_dir_id = gen_swhid(
DIRECTORY, directory_entries[0]["dir_id"], metadata=swhid_context
)
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
assert_contains(resp, "swh-take-new-snapshot")
_check_origin_link(resp, origin_info["url"])
def _origin_branches_test_helper(
client, origin_info, origin_snapshot, snapshot_id=None
):
query_params = {"origin_url": origin_info["url"], "snapshot": snapshot_id}
url = reverse("browse-origin-branches", query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/branches.html")
origin_branches = origin_snapshot[0]
origin_releases = origin_snapshot[1]
origin_branches_url = reverse("browse-origin-branches", query_params=query_params)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(origin_branches)})")
origin_releases_url = reverse("browse-origin-releases", query_params=query_params)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, f'href="{escape(origin_releases_url)}">')
assert_contains(resp, f"Releases ({nb_releases})")
assert_contains(resp, '<tr class="swh-branch-entry', count=len(origin_branches))
for branch in origin_branches:
browse_branch_url = reverse(
"browse-origin-directory",
query_params={"branch": branch["name"], **query_params},
)
assert_contains(resp, '<a href="%s">' % escape(browse_branch_url))
browse_revision_url = reverse(
"browse-revision",
url_args={"sha1_git": branch["revision"]},
query_params=query_params,
)
assert_contains(resp, '<a href="%s">' % escape(browse_revision_url))
_check_origin_link(resp, origin_info["url"])
def _origin_releases_test_helper(
client, origin_info, origin_snapshot, snapshot_id=None
):
query_params = {"origin_url": origin_info["url"], "snapshot": snapshot_id}
url = reverse("browse-origin-releases", query_params=query_params)
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/releases.html")
origin_branches = origin_snapshot[0]
origin_releases = origin_snapshot[1]
origin_branches_url = reverse("browse-origin-branches", query_params=query_params)
assert_contains(resp, f'href="{escape(origin_branches_url)}"')
assert_contains(resp, f"Branches ({len(origin_branches)})")
origin_releases_url = reverse("browse-origin-releases", query_params=query_params)
nb_releases = len(origin_releases)
if nb_releases > 0:
assert_contains(resp, f'href="{escape(origin_releases_url)}"')
assert_contains(resp, f"Releases ({nb_releases})")
assert_contains(resp, '<tr class="swh-release-entry', count=nb_releases)
for release in origin_releases:
browse_release_url = reverse(
"browse-release",
url_args={"sha1_git": release["id"]},
query_params=query_params,
)
browse_revision_url = reverse(
"browse-revision",
url_args={"sha1_git": release["target"]},
query_params=query_params,
)
assert_contains(resp, '<a href="%s">' % escape(browse_release_url))
assert_contains(resp, '<a href="%s">' % escape(browse_revision_url))
_check_origin_link(resp, origin_info["url"])
@given(
new_origin(), visit_dates(), revisions(min_size=10, max_size=10), existing_release()
)
def test_origin_branches_pagination_with_alias(
client, archive_data, mocker, new_origin, visit_dates, revisions, existing_release
):
"""
When a snapshot contains a branch or a release alias, pagination links
in the branches / releases view should be displayed.
"""
mocker.patch("swh.web.browse.snapshot_context.PER_PAGE", len(revisions) / 2)
snp_dict = {"branches": {}, "id": hash_to_bytes(random_sha1())}
for i in range(len(revisions)):
branch = "".join(random.choices(string.ascii_lowercase, k=8))
snp_dict["branches"][branch.encode()] = {
"target_type": "revision",
"target": hash_to_bytes(revisions[i]),
}
release = "".join(random.choices(string.ascii_lowercase, k=8))
snp_dict["branches"][b"RELEASE_ALIAS"] = {
"target_type": "alias",
"target": release.encode(),
}
snp_dict["branches"][release.encode()] = {
"target_type": "release",
"target": hash_to_bytes(existing_release),
}
archive_data.origin_add([new_origin])
archive_data.snapshot_add([Snapshot.from_dict(snp_dict)])
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_dates[0], type="git",)]
)[0]
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=now(),
status="full",
snapshot=snp_dict["id"],
)
archive_data.origin_visit_status_add([visit_status])
url = reverse("browse-origin-branches", query_params={"origin_url": new_origin.url})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/branches.html")
assert_contains(resp, '<ul class="pagination')
def _check_origin_link(resp, origin_url):
browse_origin_url = reverse(
"browse-origin", query_params={"origin_url": origin_url}
)
assert_contains(resp, f'href="{browse_origin_url}"')
diff --git a/swh/web/tests/browse/views/test_release.py b/swh/web/tests/browse/views/test_release.py
index d0c99eb4..de2d0f58 100644
--- a/swh/web/tests/browse/views/test_release.py
+++ b/swh/web/tests/browse/views/test_release.py
@@ -1,144 +1,145 @@
# Copyright (C) 2018-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
-from django.utils.html import escape
from hypothesis import given
+from django.utils.html import escape
+
from swh.web.common.identifiers import gen_swhid
from swh.web.common.utils import format_utc_iso_date, reverse
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.tests.strategies import origin_with_releases, release, unknown_release
@given(release())
def test_release_browse(client, archive_data, release):
_release_browse_checks(client, release, archive_data)
@given(origin_with_releases())
def test_release_browse_with_origin_snapshot(client, archive_data, origin):
snapshot = archive_data.snapshot_get_latest(origin["url"])
release = random.choice(
[
b["target"]
for b in snapshot["branches"].values()
if b["target_type"] == "release"
]
)
_release_browse_checks(client, release, archive_data, origin_url=origin["url"])
_release_browse_checks(client, release, archive_data, snapshot_id=snapshot["id"])
_release_browse_checks(
client,
release,
archive_data,
origin_url=origin["url"],
snapshot_id=snapshot["id"],
)
@given(unknown_release())
def test_release_browse_not_found(client, archive_data, unknown_release):
url = reverse("browse-release", url_args={"sha1_git": unknown_release})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
err_msg = "Release with sha1_git %s not found" % unknown_release
assert_contains(resp, err_msg, status_code=404)
@given(release())
def test_release_uppercase(client, release):
url = reverse(
"browse-release-uppercase-checksum", url_args={"sha1_git": release.upper()}
)
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse("browse-release", url_args={"sha1_git": release})
assert resp["location"] == redirect_url
def _release_browse_checks(
client, release, archive_data, origin_url=None, snapshot_id=None
):
query_params = {"origin_url": origin_url, "snapshot": snapshot_id}
url = reverse(
"browse-release", url_args={"sha1_git": release}, query_params=query_params
)
release_data = archive_data.release_get(release)
resp = client.get(url)
release_id = release_data["id"]
release_name = release_data["name"]
author_name = release_data["author"]["name"]
release_date = release_data["date"]
message = release_data["message"]
target_type = release_data["target_type"]
target = release_data["target"]
target_url = reverse(
"browse-revision", url_args={"sha1_git": target}, query_params=query_params
)
message_lines = message.split("\n")
assert resp.status_code == 200
assert_template_used(resp, "browse/release.html")
assert_contains(resp, author_name)
assert_contains(resp, format_utc_iso_date(release_date))
assert_contains(
resp,
"<h6>%s</h6>%s" % (message_lines[0] or "None", "\n".join(message_lines[1:])),
)
assert_contains(resp, release_id)
assert_contains(resp, release_name)
assert_contains(resp, target_type)
assert_contains(resp, '<a href="%s">%s</a>' % (escape(target_url), target))
swh_rel_id = gen_swhid("release", release_id)
swh_rel_id_url = reverse("browse-swhid", url_args={"swhid": swh_rel_id})
assert_contains(resp, swh_rel_id)
assert_contains(resp, swh_rel_id_url)
if origin_url:
browse_origin_url = reverse(
"browse-origin", query_params={"origin_url": origin_url}
)
assert_contains(resp, f'href="{browse_origin_url}"')
elif snapshot_id:
swh_snp_id = gen_swhid("snapshot", snapshot_id)
swh_snp_id_url = reverse("browse-swhid", url_args={"swhid": swh_snp_id})
assert_contains(resp, f'href="{swh_snp_id_url}"')
if release_data["target_type"] == "revision":
if origin_url:
directory_url = reverse(
"browse-origin-directory",
query_params={
"origin_url": origin_url,
"release": release_data["name"],
"snapshot": snapshot_id,
},
)
elif snapshot_id:
directory_url = reverse(
"browse-snapshot-directory",
url_args={"snapshot_id": snapshot_id},
query_params={"release": release_data["name"],},
)
else:
rev = archive_data.revision_get(release_data["target"])
directory_url = reverse(
"browse-directory", url_args={"sha1_git": rev["directory"]}
)
assert_contains(resp, escape(directory_url))
diff --git a/swh/web/tests/browse/views/test_revision.py b/swh/web/tests/browse/views/test_revision.py
index 9e2c494e..4445fd14 100644
--- a/swh/web/tests/browse/views/test_revision.py
+++ b/swh/web/tests/browse/views/test_revision.py
@@ -1,294 +1,295 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
-from django.utils.html import escape
from hypothesis import given
+from django.utils.html import escape
+
from swh.model.identifiers import DIRECTORY, REVISION, SNAPSHOT
from swh.web.common.identifiers import gen_swhid
from swh.web.common.utils import format_utc_iso_date, parse_iso8601_date_to_utc, reverse
from swh.web.tests.django_asserts import assert_contains, assert_template_used
from swh.web.tests.strategies import new_origin, origin, revision, unknown_revision
@given(revision())
def test_revision_browse(client, archive_data, revision):
_revision_browse_checks(client, archive_data, revision)
@given(origin())
def test_revision_origin_snapshot_browse(client, archive_data, origin):
snapshot = archive_data.snapshot_get_latest(origin["url"])
revision = archive_data.snapshot_get_head(snapshot)
_revision_browse_checks(client, archive_data, revision, origin_url=origin["url"])
_revision_browse_checks(client, archive_data, revision, snapshot=snapshot)
_revision_browse_checks(
client, archive_data, revision, origin_url=origin["url"], snapshot=snapshot,
)
revision = random.choice(archive_data.revision_log(revision))["id"]
_revision_browse_checks(client, archive_data, revision, origin_url=origin["url"])
@given(revision())
def test_revision_log_browse(client, archive_data, revision):
per_page = 10
revision_log = archive_data.revision_log(revision)
revision_log_sorted = sorted(
revision_log,
key=lambda rev: -parse_iso8601_date_to_utc(rev["committer_date"]).timestamp(),
)
url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"per_page": per_page},
)
resp = client.get(url)
next_page_url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"offset": per_page, "per_page": per_page,},
)
nb_log_entries = per_page
if len(revision_log_sorted) < per_page:
nb_log_entries = len(revision_log_sorted)
assert resp.status_code == 200
assert_template_used(resp, "browse/revision-log.html")
assert_contains(resp, '<tr class="swh-revision-log-entry', count=nb_log_entries)
assert_contains(resp, '<a class="page-link">Newer</a>')
if len(revision_log_sorted) > per_page:
assert_contains(
resp, '<a class="page-link" href="%s">Older</a>' % escape(next_page_url),
)
for log in revision_log_sorted[:per_page]:
revision_url = reverse("browse-revision", url_args={"sha1_git": log["id"]})
assert_contains(resp, log["id"][:7])
assert_contains(resp, log["author"]["name"])
assert_contains(resp, format_utc_iso_date(log["date"]))
assert_contains(resp, escape(log["message"]))
assert_contains(resp, format_utc_iso_date(log["committer_date"]))
assert_contains(resp, revision_url)
if len(revision_log_sorted) <= per_page:
return
resp = client.get(next_page_url)
prev_page_url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"offset": 0, "per_page": per_page},
)
next_page_url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"offset": 2 * per_page, "per_page": per_page},
)
nb_log_entries = len(revision_log_sorted) - per_page
if nb_log_entries > per_page:
nb_log_entries = per_page
assert resp.status_code == 200
assert_template_used(resp, "browse/revision-log.html")
assert_contains(resp, '<tr class="swh-revision-log-entry', count=nb_log_entries)
assert_contains(
resp, '<a class="page-link" href="%s">Newer</a>' % escape(prev_page_url)
)
if len(revision_log_sorted) > 2 * per_page:
assert_contains(
resp, '<a class="page-link" href="%s">Older</a>' % escape(next_page_url),
)
if len(revision_log_sorted) <= 2 * per_page:
return
resp = client.get(next_page_url)
prev_page_url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"offset": per_page, "per_page": per_page},
)
next_page_url = reverse(
"browse-revision-log",
url_args={"sha1_git": revision},
query_params={"offset": 3 * per_page, "per_page": per_page},
)
nb_log_entries = len(revision_log_sorted) - 2 * per_page
if nb_log_entries > per_page:
nb_log_entries = per_page
assert resp.status_code == 200
assert_template_used(resp, "browse/revision-log.html")
assert_contains(resp, '<tr class="swh-revision-log-entry', count=nb_log_entries)
assert_contains(
resp, '<a class="page-link" href="%s">Newer</a>' % escape(prev_page_url)
)
if len(revision_log_sorted) > 3 * per_page:
assert_contains(
resp, '<a class="page-link" href="%s">Older</a>' % escape(next_page_url),
)
@given(revision(), unknown_revision(), new_origin())
def test_revision_request_errors(client, revision, unknown_revision, new_origin):
url = reverse("browse-revision", url_args={"sha1_git": unknown_revision})
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(
resp, "Revision with sha1_git %s not found" % unknown_revision, status_code=404
)
url = reverse(
"browse-revision",
url_args={"sha1_git": revision},
query_params={"origin_url": new_origin.url},
)
resp = client.get(url)
assert resp.status_code == 404
assert_template_used(resp, "error.html")
assert_contains(
resp, "the origin mentioned in your request" " appears broken", status_code=404
)
@given(revision())
def test_revision_uppercase(client, revision):
url = reverse(
"browse-revision-uppercase-checksum", url_args={"sha1_git": revision.upper()}
)
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse("browse-revision", url_args={"sha1_git": revision})
assert resp["location"] == redirect_url
def _revision_browse_checks(
client, archive_data, revision, origin_url=None, snapshot=None
):
query_params = {}
if origin_url:
query_params["origin_url"] = origin_url
if snapshot:
query_params["snapshot"] = snapshot["id"]
url = reverse(
"browse-revision", url_args={"sha1_git": revision}, query_params=query_params
)
revision_data = archive_data.revision_get(revision)
author_name = revision_data["author"]["name"]
committer_name = revision_data["committer"]["name"]
dir_id = revision_data["directory"]
if origin_url:
snapshot = archive_data.snapshot_get_latest(origin_url)
history_url = reverse(
"browse-origin-log", query_params={"revision": revision, **query_params},
)
elif snapshot:
history_url = reverse(
"browse-snapshot-log",
url_args={"snapshot_id": snapshot["id"]},
query_params={"revision": revision},
)
else:
history_url = reverse("browse-revision-log", url_args={"sha1_git": revision})
resp = client.get(url)
assert resp.status_code == 200
assert_template_used(resp, "browse/revision.html")
assert_contains(resp, author_name)
assert_contains(resp, committer_name)
assert_contains(resp, history_url)
for parent in revision_data["parents"]:
parent_url = reverse(
"browse-revision", url_args={"sha1_git": parent}, query_params=query_params
)
assert_contains(resp, '<a href="%s">%s</a>' % (escape(parent_url), parent[:7]))
author_date = revision_data["date"]
committer_date = revision_data["committer_date"]
message_lines = revision_data["message"].split("\n")
assert_contains(resp, format_utc_iso_date(author_date))
assert_contains(resp, format_utc_iso_date(committer_date))
assert_contains(resp, escape(message_lines[0]))
assert_contains(resp, escape("\n".join(message_lines[1:])))
assert_contains(resp, "vault-cook-directory")
assert_contains(resp, "vault-cook-revision")
swh_rev_id = gen_swhid("revision", revision)
swh_rev_id_url = reverse("browse-swhid", url_args={"swhid": swh_rev_id})
assert_contains(resp, swh_rev_id)
assert_contains(resp, swh_rev_id_url)
swh_dir_id = gen_swhid("directory", dir_id)
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
if origin_url:
assert_contains(resp, "swh-take-new-snapshot")
swh_rev_id = gen_swhid(REVISION, revision)
swh_rev_id_url = reverse("browse-swhid", url_args={"swhid": swh_rev_id})
if origin_url:
browse_origin_url = reverse(
"browse-origin", query_params={"origin_url": origin_url}
)
assert_contains(resp, f'href="{browse_origin_url}"')
elif snapshot:
swh_snp_id = gen_swhid("snapshot", snapshot["id"])
swh_snp_id_url = reverse("browse-swhid", url_args={"swhid": swh_snp_id})
assert_contains(resp, f'href="{swh_snp_id_url}"')
swhid_context = {}
if origin_url:
swhid_context["origin"] = origin_url
if snapshot:
swhid_context["visit"] = gen_swhid(SNAPSHOT, snapshot["id"])
swh_rev_id = gen_swhid(REVISION, revision, metadata=swhid_context)
swh_rev_id_url = reverse("browse-swhid", url_args={"swhid": swh_rev_id})
assert_contains(resp, swh_rev_id)
assert_contains(resp, swh_rev_id_url)
swhid_context["anchor"] = gen_swhid(REVISION, revision)
swhid_context["path"] = "/"
swh_dir_id = gen_swhid(DIRECTORY, dir_id, metadata=swhid_context)
swh_dir_id_url = reverse("browse-swhid", url_args={"swhid": swh_dir_id})
assert_contains(resp, swh_dir_id)
assert_contains(resp, swh_dir_id_url)
diff --git a/swh/web/tests/common/test_utils.py b/swh/web/tests/common/test_utils.py
index 360cd628..c57b8de1 100644
--- a/swh/web/tests/common/test_utils.py
+++ b/swh/web/tests/common/test_utils.py
@@ -1,234 +1,235 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import datetime
from urllib.parse import quote
+import pytest
+
from django.conf.urls import url
from django.test.utils import override_settings
from django.urls.exceptions import NoReverseMatch
-import pytest
from swh.web.common import utils
from swh.web.common.exc import BadInputExc
def test_shorten_path_noop():
noops = ["/api/", "/browse/", "/content/symbol/foobar/"]
for noop in noops:
assert utils.shorten_path(noop) == noop
def test_shorten_path_sha1():
sha1 = "aafb16d69fd30ff58afdd69036a26047f3aebdc6"
short_sha1 = sha1[:8] + "..."
templates = [
"/api/1/content/sha1:%s/",
"/api/1/content/sha1_git:%s/",
"/api/1/directory/%s/",
"/api/1/content/sha1:%s/ctags/",
]
for template in templates:
assert utils.shorten_path(template % sha1) == template % short_sha1
def test_shorten_path_sha256():
sha256 = "aafb16d69fd30ff58afdd69036a26047" "213add102934013a014dfca031c41aef"
short_sha256 = sha256[:8] + "..."
templates = [
"/api/1/content/sha256:%s/",
"/api/1/directory/%s/",
"/api/1/content/sha256:%s/filetype/",
]
for template in templates:
assert utils.shorten_path(template % sha256) == template % short_sha256
@pytest.mark.parametrize(
"input_timestamp, output_date",
[
(
"2016-01-12",
datetime.datetime(2016, 1, 12, 0, 0, tzinfo=datetime.timezone.utc),
),
(
"2016-01-12T09:19:12+0100",
datetime.datetime(2016, 1, 12, 8, 19, 12, tzinfo=datetime.timezone.utc),
),
(
"2007-01-14T20:34:22Z",
datetime.datetime(2007, 1, 14, 20, 34, 22, tzinfo=datetime.timezone.utc),
),
],
)
def test_parse_iso8601_date_to_utc_ok(input_timestamp, output_date):
assert utils.parse_iso8601_date_to_utc(input_timestamp) == output_date
@pytest.mark.parametrize(
"invalid_iso8601_timestamp", ["Today is January 1, 2047 at 8:21:00AM", "1452591542"]
)
def test_parse_iso8601_date_to_utc_ko(invalid_iso8601_timestamp):
with pytest.raises(BadInputExc):
utils.parse_iso8601_date_to_utc(invalid_iso8601_timestamp)
def test_format_utc_iso_date():
assert (
utils.format_utc_iso_date("2017-05-04T13:27:13+02:00")
== "04 May 2017, 11:27 UTC"
)
def test_gen_path_info():
input_path = "/home/user/swh-environment/swh-web/"
expected_result = [
{"name": "home", "path": "home"},
{"name": "user", "path": "home/user"},
{"name": "swh-environment", "path": "home/user/swh-environment"},
{"name": "swh-web", "path": "home/user/swh-environment/swh-web"},
]
path_info = utils.gen_path_info(input_path)
assert path_info == expected_result
input_path = "home/user/swh-environment/swh-web"
path_info = utils.gen_path_info(input_path)
assert path_info == expected_result
def test_rst_to_html():
rst = (
"Section\n"
"=======\n\n"
"**Some strong text**\n\n"
"Subsection\n"
"----------\n\n"
"* This is a bulleted list.\n"
"* It has two items, the second\n"
" item uses two lines.\n"
"\n"
"1. This is a numbered list.\n"
"2. It has two items too.\n"
"\n"
"#. This is a numbered list.\n"
"#. It has two items too.\n"
)
expected_html = (
'<div class="swh-rst"><h1 class="title">Section</h1>\n'
"<p><strong>Some strong text</strong></p>\n"
'<div class="section" id="subsection">\n'
"<h2>Subsection</h2>\n"
'<ul class="simple">\n'
"<li><p>This is a bulleted list.</p></li>\n"
"<li><p>It has two items, the second\n"
"item uses two lines.</p></li>\n"
"</ul>\n"
'<ol class="arabic simple">\n'
"<li><p>This is a numbered list.</p></li>\n"
"<li><p>It has two items too.</p></li>\n"
"<li><p>This is a numbered list.</p></li>\n"
"<li><p>It has two items too.</p></li>\n"
"</ol>\n"
"</div>\n"
"</div>"
)
assert utils.rst_to_html(rst) == expected_html
def sample_test_view(request, string, number):
pass
def sample_test_view_no_url_args(request):
pass
urlpatterns = [
url(
r"^sample/test/(?P<string>.+)/view/(?P<number>[0-9]+)/$",
sample_test_view,
name="sample-test-view",
),
url(
r"^sample/test/view/no/url/args/$",
sample_test_view_no_url_args,
name="sample-test-view-no-url-args",
),
]
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_url_args_only_ok():
string = "foo"
number = 55
url = utils.reverse(
"sample-test-view", url_args={"string": string, "number": number}
)
assert url == f"/sample/test/{string}/view/{number}/"
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_url_args_only_ko():
string = "foo"
with pytest.raises(NoReverseMatch):
utils.reverse("sample-test-view", url_args={"string": string, "number": string})
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_no_url_args():
url = utils.reverse("sample-test-view-no-url-args")
assert url == "/sample/test/view/no/url/args/"
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_query_params_only():
start = 0
scope = "foo"
url = utils.reverse(
"sample-test-view-no-url-args", query_params={"start": start, "scope": scope}
)
assert url == f"/sample/test/view/no/url/args/?scope={scope}&start={start}"
url = utils.reverse(
"sample-test-view-no-url-args", query_params={"start": start, "scope": None}
)
assert url == f"/sample/test/view/no/url/args/?start={start}"
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_query_params_encode():
libname = "libstc++"
url = utils.reverse(
"sample-test-view-no-url-args", query_params={"libname": libname}
)
assert url == f"/sample/test/view/no/url/args/?libname={quote(libname, safe='/;:')}"
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_url_args_query_params():
string = "foo"
number = 55
start = 10
scope = "bar"
url = utils.reverse(
"sample-test-view",
url_args={"string": string, "number": number},
query_params={"start": start, "scope": scope},
)
assert url == f"/sample/test/{string}/view/{number}/?scope={scope}&start={start}"
@override_settings(ROOT_URLCONF=__name__)
def test_reverse_absolute_uri(request_factory):
request = request_factory.get(utils.reverse("sample-test-view-no-url-args"))
url = utils.reverse("sample-test-view-no-url-args", request=request)
assert url == f"http://{request.META['SERVER_NAME']}/sample/test/view/no/url/args/"
diff --git a/swh/web/tests/conftest.py b/swh/web/tests/conftest.py
index 4ef9fa0a..c14d2741 100644
--- a/swh/web/tests/conftest.py
+++ b/swh/web/tests/conftest.py
@@ -1,360 +1,361 @@
# Copyright (C) 2018-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import json
import os
import shutil
from subprocess import PIPE, run
import sys
from typing import Any, Dict, List, Optional
-from django.core.cache import cache
from hypothesis import HealthCheck, settings
import pytest
+
+from django.core.cache import cache
from rest_framework.test import APIClient, APIRequestFactory
from swh.model.hashutil import ALGORITHMS, hash_to_bytes
from swh.storage.algos.origin import origin_get_latest_visit_status
from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest
from swh.web.common import converters
from swh.web.common.typing import OriginVisitInfo
from swh.web.tests.data import get_tests_data, override_storages
# Used to skip some tests
ctags_json_missing = (
shutil.which("ctags") is None
or b"+json" not in run(["ctags", "--version"], stdout=PIPE).stdout
)
fossology_missing = shutil.which("nomossa") is None
# Register some hypothesis profiles
settings.register_profile("default", settings())
settings.register_profile(
"swh-web",
settings(
deadline=None,
suppress_health_check=[HealthCheck.too_slow, HealthCheck.filter_too_much],
),
)
settings.register_profile(
"swh-web-fast",
settings(
deadline=None,
max_examples=1,
suppress_health_check=[HealthCheck.too_slow, HealthCheck.filter_too_much],
),
)
def pytest_configure(config):
# Use fast hypothesis profile by default if none has been
# explicitly specified in pytest option
if config.getoption("--hypothesis-profile") is None:
settings.load_profile("swh-web-fast")
# Small hack in order to be able to run the unit tests
# without static assets generated by webpack.
# Those assets are not really needed for the Python tests
# but the django templates will fail to load due to missing
# generated file webpack-stats.json describing the js and css
# files to include.
# So generate a dummy webpack-stats.json file to overcome
# that issue.
test_dir = os.path.dirname(__file__)
# location of the static folder when running tests through tox
static_dir = os.path.join(sys.prefix, "share/swh/web/static")
if not os.path.exists(static_dir):
# location of the static folder when running tests locally with pytest
static_dir = os.path.join(test_dir, "../../../static")
webpack_stats = os.path.join(static_dir, "webpack-stats.json")
if os.path.exists(webpack_stats):
return
bundles_dir = os.path.join(test_dir, "../assets/src/bundles")
_, dirs, _ = next(os.walk(bundles_dir))
mock_webpack_stats = {"status": "done", "publicPath": "/static", "chunks": {}}
for bundle in dirs:
asset = "js/%s.js" % bundle
mock_webpack_stats["chunks"][bundle] = [
{
"name": asset,
"publicPath": "/static/%s" % asset,
"path": os.path.join(static_dir, asset),
}
]
with open(webpack_stats, "w") as outfile:
json.dump(mock_webpack_stats, outfile)
# Clear Django cache before each test
@pytest.fixture(autouse=True)
def django_cache_cleared():
cache.clear()
# Alias rf fixture from pytest-django
@pytest.fixture
def request_factory(rf):
return rf
# Fixture to get test client from Django REST Framework
@pytest.fixture(scope="module")
def api_client():
return APIClient()
# Fixture to get API request factory from Django REST Framework
@pytest.fixture(scope="module")
def api_request_factory():
return APIRequestFactory()
# Initialize tests data
@pytest.fixture(scope="session", autouse=True)
def tests_data():
data = get_tests_data(reset=True)
# Update swh-web configuration to use the in-memory storages
# instantiated in the tests.data module
override_storages(data["storage"], data["idx_storage"], data["search"])
return data
# Fixture to manipulate data from a sample archive used in the tests
@pytest.fixture(scope="session")
def archive_data(tests_data):
return _ArchiveData(tests_data)
# Fixture to manipulate indexer data from a sample archive used in the tests
@pytest.fixture(scope="session")
def indexer_data(tests_data):
return _IndexerData(tests_data)
# Custom data directory for requests_mock
@pytest.fixture
def datadir():
return os.path.join(os.path.abspath(os.path.dirname(__file__)), "resources")
class _ArchiveData:
"""
Helper class to manage data from a sample test archive.
It is initialized with a reference to an in-memory storage
containing raw tests data.
It is basically a proxy to Storage interface but it overrides some methods
to retrieve those tests data in a json serializable format in order to ease
tests implementation.
"""
def __init__(self, tests_data):
self.storage = tests_data["storage"]
def __getattr__(self, key):
if key == "storage":
raise AttributeError(key)
# Forward calls to non overridden Storage methods to wrapped
# storage instance
return getattr(self.storage, key)
def content_find(self, content: Dict[str, Any]) -> Dict[str, Any]:
cnt_ids_bytes = {
algo_hash: hash_to_bytes(content[algo_hash])
for algo_hash in ALGORITHMS
if content.get(algo_hash)
}
cnt = self.storage.content_find(cnt_ids_bytes)
return converters.from_content(cnt[0].to_dict()) if cnt else cnt
def content_get(self, cnt_id: str) -> Dict[str, Any]:
cnt_id_bytes = hash_to_bytes(cnt_id)
content = self.storage.content_get([cnt_id_bytes])[0]
if content:
content_d = content.to_dict()
content_d.pop("ctime", None)
else:
content_d = None
return converters.from_swh(
content_d, hashess={"sha1", "sha1_git", "sha256", "blake2s256"}
)
def content_get_data(self, cnt_id: str) -> Optional[Dict[str, Any]]:
cnt_id_bytes = hash_to_bytes(cnt_id)
cnt_data = self.storage.content_get_data(cnt_id_bytes)
if cnt_data is None:
return None
return converters.from_content({"data": cnt_data, "sha1": cnt_id_bytes})
def directory_get(self, dir_id):
return {"id": dir_id, "content": self.directory_ls(dir_id)}
def directory_ls(self, dir_id):
cnt_id_bytes = hash_to_bytes(dir_id)
dir_content = map(
converters.from_directory_entry, self.storage.directory_ls(cnt_id_bytes)
)
return list(dir_content)
def release_get(self, rel_id: str) -> Optional[Dict[str, Any]]:
rel_id_bytes = hash_to_bytes(rel_id)
rel_data = self.storage.release_get([rel_id_bytes])[0]
return converters.from_release(rel_data) if rel_data else None
def revision_get(self, rev_id: str) -> Optional[Dict[str, Any]]:
rev_id_bytes = hash_to_bytes(rev_id)
rev_data = self.storage.revision_get([rev_id_bytes])[0]
return converters.from_revision(rev_data) if rev_data else None
def revision_log(self, rev_id, limit=None):
rev_id_bytes = hash_to_bytes(rev_id)
return list(
map(
converters.from_revision,
self.storage.revision_log([rev_id_bytes], limit=limit),
)
)
def snapshot_get_latest(self, origin_url):
snp = snapshot_get_latest(self.storage, origin_url)
return converters.from_snapshot(snp.to_dict())
def origin_get(self, origin_urls):
origins = self.storage.origin_get(origin_urls)
return [converters.from_origin(o.to_dict()) for o in origins]
def origin_visit_get(self, origin_url):
next_page_token = None
visits = []
while True:
visit_page = self.storage.origin_visit_get(
origin_url, page_token=next_page_token
)
next_page_token = visit_page.next_page_token
for visit in visit_page.results:
visit_status = self.storage.origin_visit_status_get_latest(
origin_url, visit.visit
)
visits.append(
converters.from_origin_visit(
{**visit_status.to_dict(), "type": visit.type}
)
)
if not next_page_token:
break
return visits
def origin_visit_get_by(self, origin_url: str, visit_id: int) -> OriginVisitInfo:
visit = self.storage.origin_visit_get_by(origin_url, visit_id)
assert visit is not None
visit_status = self.storage.origin_visit_status_get_latest(origin_url, visit_id)
assert visit_status is not None
return converters.from_origin_visit(
{**visit_status.to_dict(), "type": visit.type}
)
def origin_visit_status_get_latest(
self,
origin_url,
type: Optional[str] = None,
allowed_statuses: Optional[List[str]] = None,
require_snapshot: bool = False,
):
visit_and_status = origin_get_latest_visit_status(
self.storage,
origin_url,
type=type,
allowed_statuses=allowed_statuses,
require_snapshot=require_snapshot,
)
return (
converters.from_origin_visit(
{**visit_and_status[0].to_dict(), **visit_and_status[1].to_dict()}
)
if visit_and_status
else None
)
def snapshot_get(self, snapshot_id):
snp = snapshot_get_all_branches(self.storage, hash_to_bytes(snapshot_id))
return converters.from_snapshot(snp.to_dict())
def snapshot_get_branches(
self, snapshot_id, branches_from="", branches_count=1000, target_types=None
):
partial_branches = self.storage.snapshot_get_branches(
hash_to_bytes(snapshot_id),
branches_from.encode(),
branches_count,
target_types,
)
return converters.from_partial_branches(partial_branches)
def snapshot_get_head(self, snapshot):
if snapshot["branches"]["HEAD"]["target_type"] == "alias":
target = snapshot["branches"]["HEAD"]["target"]
head = snapshot["branches"][target]["target"]
else:
head = snapshot["branches"]["HEAD"]["target"]
return head
class _IndexerData:
"""
Helper class to manage indexer tests data
It is initialized with a reference to an in-memory indexer storage
containing raw tests data.
It also defines class methods to retrieve those tests data in
a json serializable format in order to ease tests implementation.
"""
def __init__(self, tests_data):
self.idx_storage = tests_data["idx_storage"]
self.mimetype_indexer = tests_data["mimetype_indexer"]
self.license_indexer = tests_data["license_indexer"]
self.ctags_indexer = tests_data["ctags_indexer"]
def content_add_mimetype(self, cnt_id):
self.mimetype_indexer.run([hash_to_bytes(cnt_id)], "update-dups")
def content_get_mimetype(self, cnt_id):
mimetype = next(self.idx_storage.content_mimetype_get([hash_to_bytes(cnt_id)]))
return converters.from_filetype(mimetype)
def content_add_language(self, cnt_id):
raise NotImplementedError("Language indexer is disabled.")
self.language_indexer.run([hash_to_bytes(cnt_id)], "update-dups")
def content_get_language(self, cnt_id):
lang = next(self.idx_storage.content_language_get([hash_to_bytes(cnt_id)]))
return converters.from_swh(lang, hashess={"id"})
def content_add_license(self, cnt_id):
self.license_indexer.run([hash_to_bytes(cnt_id)], "update-dups")
def content_get_license(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
lic = next(self.idx_storage.content_fossology_license_get([cnt_id_bytes]))
return converters.from_swh(
{"id": cnt_id_bytes, "facts": lic[cnt_id_bytes]}, hashess={"id"}
)
def content_add_ctags(self, cnt_id):
self.ctags_indexer.run([hash_to_bytes(cnt_id)], "update-dups")
def content_get_ctags(self, cnt_id):
cnt_id_bytes = hash_to_bytes(cnt_id)
ctags = self.idx_storage.content_ctags_get([cnt_id_bytes])
for ctag in ctags:
yield converters.from_swh(ctag, hashess={"id"})
diff --git a/swh/web/tests/misc/test_origin_save.py b/swh/web/tests/misc/test_origin_save.py
index 90b46745..41531cf4 100644
--- a/swh/web/tests/misc/test_origin_save.py
+++ b/swh/web/tests/misc/test_origin_save.py
@@ -1,105 +1,106 @@
# Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import datetime
-from django.test import Client
import pytest
+from django.test import Client
+
from swh.web.common.origin_save import (
SAVE_REQUEST_ACCEPTED,
SAVE_TASK_NOT_YET_SCHEDULED,
)
from swh.web.common.utils import reverse
from swh.web.settings.tests import save_origin_rate_post
visit_type = "git"
origin = {"url": "https://github.com/python/cpython"}
@pytest.fixture
def client():
return Client(enforce_csrf_checks=True)
def test_save_request_form_csrf_token(client, mocker):
mock_create_save_origin_request = mocker.patch(
"swh.web.misc.origin_save.create_save_origin_request"
)
_mock_create_save_origin_request(mock_create_save_origin_request)
url = reverse(
"origin-save-request",
url_args={"visit_type": visit_type, "origin_url": origin["url"]},
)
resp = client.post(url)
assert resp.status_code == 403
data = _get_csrf_token(client, reverse("origin-save"))
resp = client.post(url, data=data)
assert resp.status_code == 200
def test_save_request_form_rate_limit(client, mocker):
mock_create_save_origin_request = mocker.patch(
"swh.web.misc.origin_save.create_save_origin_request"
)
_mock_create_save_origin_request(mock_create_save_origin_request)
url = reverse(
"origin-save-request",
url_args={"visit_type": visit_type, "origin_url": origin["url"]},
)
data = _get_csrf_token(client, reverse("origin-save"))
for _ in range(save_origin_rate_post):
resp = client.post(url, data=data)
assert resp.status_code == 200
resp = client.post(url, data=data)
assert resp.status_code == 429
def test_save_request_form_server_error(client, mocker):
mock_create_save_origin_request = mocker.patch(
"swh.web.misc.origin_save.create_save_origin_request"
)
mock_create_save_origin_request.side_effect = Exception("Server error")
url = reverse(
"origin-save-request",
url_args={"visit_type": visit_type, "origin_url": origin["url"]},
)
data = _get_csrf_token(client, reverse("origin-save"))
resp = client.post(url, data=data)
assert resp.status_code == 500
def test_old_save_url_redirection(client):
url = reverse("browse-origin-save")
resp = client.get(url)
assert resp.status_code == 302
redirect_url = reverse("origin-save")
assert resp["location"] == redirect_url
def _get_csrf_token(client, url):
resp = client.get(url)
return {"csrfmiddlewaretoken": resp.cookies["csrftoken"].value}
def _mock_create_save_origin_request(mock):
expected_data = {
"visit_type": visit_type,
"origin_url": origin["url"],
"save_request_date": datetime.now().isoformat(),
"save_request_status": SAVE_REQUEST_ACCEPTED,
"save_task_status": SAVE_TASK_NOT_YET_SCHEDULED,
"visit_date": None,
}
mock.return_value = expected_data
diff --git a/swh/web/urls.py b/swh/web/urls.py
index 47517b6d..fba69b77 100644
--- a/swh/web/urls.py
+++ b/swh/web/urls.py
@@ -1,75 +1,76 @@
# Copyright (C) 2017-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
+from django_js_reverse.views import urls_js
+
from django.conf import settings
from django.conf.urls import (
handler400,
handler403,
handler404,
handler500,
include,
url,
)
from django.contrib.auth.views import LogoutView
from django.contrib.staticfiles.views import serve
from django.shortcuts import render
from django.views.generic.base import RedirectView
-from django_js_reverse.views import urls_js
from swh.web.browse.identifiers import swhid_browse
from swh.web.common.exc import (
swh_handle400,
swh_handle403,
swh_handle404,
swh_handle500,
)
from swh.web.config import get_config
swh_web_config = get_config()
favicon_view = RedirectView.as_view(
url="/static/img/icons/swh-logo-32x32.png", permanent=True
)
def _default_view(request):
return render(request, "homepage.html")
urlpatterns = [
url(r"^admin/", include("swh.web.admin.urls")),
url(r"^favicon\.ico$", favicon_view),
url(r"^api/", include("swh.web.api.urls")),
url(r"^browse/", include("swh.web.browse.urls")),
url(r"^$", _default_view, name="swh-web-homepage"),
url(r"^jsreverse/$", urls_js, name="js_reverse"),
url(
r"^(?P<swhid>swh:[0-9]+:[a-z]+:[0-9a-f]+.*)/$",
swhid_browse,
name="browse-swhid",
),
url(r"^", include("swh.web.misc.urls")),
url(r"^", include("swh.web.auth.views")),
url(r"^logout/$", LogoutView.as_view(template_name="logout.html"), name="logout"),
]
# allow to serve assets through django staticfiles
# even if settings.DEBUG is False
def insecure_serve(request, path, **kwargs):
return serve(request, path, insecure=True, **kwargs)
# enable to serve compressed assets through django development server
if swh_web_config["serve_assets"]:
static_pattern = r"^%s(?P<path>.*)$" % settings.STATIC_URL[1:]
urlpatterns.append(url(static_pattern, insecure_serve))
handler400 = swh_handle400 # noqa
handler403 = swh_handle403 # noqa
handler404 = swh_handle404 # noqa
handler500 = swh_handle500 # noqa